diff --git a/CLAUDE.md b/CLAUDE.md index a1ed95f..e017e0f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -137,9 +137,6 @@ transcriber, which owns the connection pool. a worker thread. The request sets `Content-Length` when the size is known. A large file never blocks the loop and never loads fully into memory. -**LeMUR** is sync-only. An `AsyncTranscript` works as a `LemurSource`, but the LeMUR call -blocks. Run it off the loop, for example with `asyncio.to_thread`. - ## Sync transcription (pre-recorded, single request) `SyncTranscriber` posts a whole audio file and returns the finished transcript in one @@ -167,7 +164,7 @@ pass a path/bytes or use `Transcriber` for URL ingestion. ```python config = aai.SyncTranscriptionConfig( prompt="Transcribe verbatim. Preserve disfluencies.", # max 4096 chars - keyterms_prompt=["AssemblyAI", "Lemur", "U3-Pro"], # max 2048 chars total + keyterms_prompt=["AssemblyAI", "Universal", "U3-Pro"], # max 2048 chars total language_codes=["es"], # or e.g. ["en", "es"] for multilingual; defaults to English ) result = aai.SyncTranscriber().transcribe("./call.wav", config=config) diff --git a/README.md b/README.md index 85bc760..3f1fdc9 100644 --- a/README.md +++ b/README.md @@ -423,7 +423,7 @@ aai.settings.api_key = "" config = aai.SyncTranscriptionConfig( prompt="Transcribe verbatim. Preserve disfluencies.", # max 4096 chars - keyterms_prompt=["AssemblyAI", "Lemur", "U3-Pro"], # max 2048 chars total + keyterms_prompt=["AssemblyAI", "Universal", "U3-Pro"], # max 2048 chars total conversation_context=[ # prior turns from the same conversation, oldest first "I'd like to book a flight to Denver.", @@ -1259,7 +1259,7 @@ aai.settings.polling_interval = 10.0 ## Playground -Visit our Playground to try our all of our Speech AI models and LeMUR for free: +Visit our Playground to try our all of our Speech AI models for free: - [Playground](https://www.assemblyai.com/dashboard/playground/) @@ -1355,8 +1355,6 @@ Notes: work at `max_concurrency`, which defaults to 8. - Neither group method drops a failure. Either the first error is raised, or you pass `return_failures=True` and get `(transcripts, errors)`. -- LeMUR is sync-only. An `AsyncTranscript` works as a `LemurSource`, but the LeMUR call - blocks. Run it in a thread, for example with `asyncio.to_thread`. For real-time streaming, use `assemblyai.streaming.v3.AsyncStreamingClient`. diff --git a/assemblyai/__init__.py b/assemblyai/__init__.py index c40bbac..7acffa6 100644 --- a/assemblyai/__init__.py +++ b/assemblyai/__init__.py @@ -2,7 +2,6 @@ from .__version__ import __version__ from .async_client import AsyncClient from .client import Client -from .lemur import Lemur from .prerecorded.v2 import AsyncTranscriber, AsyncTranscript from .sync import SyncTranscriber from .transcriber import Transcriber, Transcript, TranscriptGroup @@ -25,21 +24,6 @@ KeytermsPromptOptions, LanguageCode, LanguageDetectionOptions, - LemurActionItemsResponse, - LemurError, - LemurModel, - LemurPurgeRequest, - LemurPurgeResponse, - LemurQuestion, - LemurQuestionAnswer, - LemurQuestionResponse, - LemurSource, - LemurSourceType, - LemurStringResponse, - LemurSummaryResponse, - LemurTaskResponse, - LemurTranscriptSource, - LemurUsage, ListTranscriptParameters, ListTranscriptResponse, PageDetails, @@ -114,22 +98,6 @@ "KeytermsPromptOptions", "LanguageCode", "LanguageDetectionOptions", - "Lemur", - "LemurActionItemsResponse", - "LemurError", - "LemurModel", - "LemurPurgeRequest", - "LemurPurgeResponse", - "LemurSource", - "LemurSourceType", - "LemurTranscriptSource", - "LemurQuestion", - "LemurQuestionAnswer", - "LemurQuestionResponse", - "LemurStringResponse", - "LemurSummaryResponse", - "LemurTaskResponse", - "LemurUsage", "ListTranscriptParameters", "ListTranscriptResponse", "PageDetails", diff --git a/assemblyai/__version__.py b/assemblyai/__version__.py index 54d5a93..b422ee5 100644 --- a/assemblyai/__version__.py +++ b/assemblyai/__version__.py @@ -1 +1 @@ -__version__ = "0.65.00" +__version__ = "0.66.00" diff --git a/assemblyai/api.py b/assemblyai/api.py index 25ad9f0..da66ea9 100644 --- a/assemblyai/api.py +++ b/assemblyai/api.py @@ -1,12 +1,10 @@ -from typing import BinaryIO, Optional, Union +from typing import BinaryIO, Union import httpx from . import types ENDPOINT_UPLOAD = "/v2/upload" -ENDPOINT_LEMUR_BASE = "/lemur/v3" -ENDPOINT_LEMUR = f"{ENDPOINT_LEMUR_BASE}/generate" def _get_error_message(response: httpx.Response) -> str: @@ -54,140 +52,6 @@ def upload_file( return response.json()["upload_url"] -def lemur_question( - client: httpx.Client, - request: types.LemurQuestionRequest, - http_timeout: Optional[float], -) -> types.LemurQuestionResponse: - response = client.post( - f"{ENDPOINT_LEMUR}/question-answer", - json=request.dict( - exclude_none=True, - ), - timeout=http_timeout, - ) - - if response.status_code != httpx.codes.OK: - raise types.LemurError( - f"failed to call Lemur questions: {_get_error_message(response)}", - response.status_code, - ) - - return types.LemurQuestionResponse.parse_obj(response.json()) - - -def lemur_summarize( - client: httpx.Client, - request: types.LemurSummaryRequest, - http_timeout: Optional[float], -) -> types.LemurSummaryResponse: - response = client.post( - f"{ENDPOINT_LEMUR}/summary", - json=request.dict( - exclude_none=True, - ), - timeout=http_timeout, - ) - - if response.status_code != httpx.codes.OK: - raise types.LemurError( - f"failed to call Lemur summary: {_get_error_message(response)}", - response.status_code, - ) - - return types.LemurSummaryResponse.parse_obj(response.json()) - - -def lemur_action_items( - client: httpx.Client, - request: types.LemurActionItemsRequest, - http_timeout: Optional[float], -) -> types.LemurActionItemsResponse: - response = client.post( - f"{ENDPOINT_LEMUR}/action-items", - json=request.dict( - exclude_none=True, - ), - timeout=http_timeout, - ) - - if response.status_code != httpx.codes.OK: - raise types.LemurError( - f"failed to call Lemur action items: {_get_error_message(response)}", - response.status_code, - ) - - return types.LemurActionItemsResponse.parse_obj(response.json()) - - -def lemur_task( - client: httpx.Client, - request: types.LemurTaskRequest, - http_timeout: Optional[float], -) -> types.LemurTaskResponse: - response = client.post( - f"{ENDPOINT_LEMUR}/task", - json=request.dict( - exclude_none=True, - ), - timeout=http_timeout, - ) - - if response.status_code != httpx.codes.OK: - raise types.LemurError( - f"failed to call Lemur task: {_get_error_message(response)}", - response.status_code, - ) - - return types.LemurTaskResponse.parse_obj(response.json()) - - -def lemur_purge_request_data( - client: httpx.Client, - request: types.LemurPurgeRequest, - http_timeout: Optional[float], -) -> types.LemurPurgeResponse: - response = client.delete( - f"{ENDPOINT_LEMUR_BASE}/{request.request_id}", - timeout=http_timeout, - ) - - if response.status_code != httpx.codes.OK: - raise types.LemurError( - f"Failed to purge LeMUR request data for provided request ID: {request.request_id}. Error: {_get_error_message(response)}", - response.status_code, - ) - - return types.LemurPurgeResponse.parse_obj(response.json()) - - -def lemur_get_response_data( - client: httpx.Client, - request_id: str, - http_timeout: Optional[float], -) -> Union[ - types.LemurStringResponse, - types.LemurQuestionResponse, -]: - response = client.get( - f"{ENDPOINT_LEMUR_BASE}/{request_id}", - timeout=http_timeout, - ) - - if response.status_code != httpx.codes.OK: - raise types.LemurError( - f"Failed to get LeMUR response data for provided request ID: {request_id}. Error: {_get_error_message(response)}", - response.status_code, - ) - - json_data = response.json() - - if isinstance(json_data.get("response"), list): - return types.LemurQuestionResponse.parse_obj(json_data) - - return types.LemurStringResponse.parse_obj(json_data) - - # Canonical location for the prerecorded transcript endpoints is # ``assemblyai.prerecorded.v2.api``. from .prerecorded.v2.api import ( # noqa: E402, F401 diff --git a/assemblyai/lemur.py b/assemblyai/lemur.py deleted file mode 100644 index affdaf8..0000000 --- a/assemblyai/lemur.py +++ /dev/null @@ -1,598 +0,0 @@ -from __future__ import annotations - -import concurrent.futures -from typing import Any, Dict, List, Optional, Union - -from . import api, types -from . import client as _client - - -class _LemurImpl: - def __init__( - self, - *, - client: _client.Client, - sources: Optional[List[types.LemurSource]], - ) -> None: - self._client = client - - self._sources = ( - [types.LemurSourceRequest.from_lemur_source(s) for s in sources] - if sources is not None - else [] - ) - - def question( - self, - questions: List[types.LemurQuestion], - context: Optional[Union[str, Dict[str, Any]]], - timeout: Optional[float], - final_model: Optional[types.LemurModel], - max_output_size: Optional[int], - temperature: Optional[float], - input_text: Optional[str], - ) -> types.LemurQuestionResponse: - response = api.lemur_question( - client=self._client.http_client, - request=types.LemurQuestionRequest( - sources=self._sources, - questions=questions, - context=context, - final_model=final_model, - max_output_size=max_output_size, - temperature=temperature, - input_text=input_text, - ), - http_timeout=timeout, - ) - - return response - - def summarize( - self, - context: Optional[Union[str, Dict[str, Any]]], - answer_format: Optional[str], - final_model: Optional[types.LemurModel], - max_output_size: Optional[int], - timeout: Optional[float], - temperature: Optional[float], - input_text: Optional[str], - ) -> types.LemurSummaryResponse: - response = api.lemur_summarize( - client=self._client.http_client, - request=types.LemurSummaryRequest( - sources=self._sources, - context=context, - answer_format=answer_format, - final_model=final_model, - max_output_size=max_output_size, - temperature=temperature, - input_text=input_text, - ), - http_timeout=timeout, - ) - - return response - - def action_items( - self, - context: Optional[Union[str, Dict[str, Any]]], - answer_format: Optional[str], - final_model: Optional[types.LemurModel], - max_output_size: Optional[int], - timeout: Optional[float], - temperature: Optional[float], - input_text: Optional[str], - ) -> types.LemurActionItemsResponse: - response = api.lemur_action_items( - client=self._client.http_client, - request=types.LemurActionItemsRequest( - sources=self._sources, - context=context, - answer_format=answer_format, - final_model=final_model, - max_output_size=max_output_size, - temperature=temperature, - input_text=input_text, - ), - http_timeout=timeout, - ) - - return response - - def task( - self, - prompt: str, - context: Optional[Union[str, Dict[str, Any]]], - final_model: Optional[types.LemurModel], - max_output_size: Optional[int], - timeout: Optional[float], - temperature: Optional[float], - input_text: Optional[str], - ): - response = api.lemur_task( - client=self._client.http_client, - request=types.LemurTaskRequest( - sources=self._sources, - prompt=prompt, - context=context, - final_model=final_model, - max_output_size=max_output_size, - temperature=temperature, - input_text=input_text, - ), - http_timeout=timeout, - ) - - return response - - @classmethod - def purge_request_data( - cls, - request_id: str, - timeout: Optional[float] = None, - ) -> types.LemurPurgeResponse: - response = api.lemur_purge_request_data( - client=_client.Client.get_default().http_client, - request=types.LemurPurgeRequest( - request_id=request_id, - ), - http_timeout=timeout, - ) - - return response - - def get_response_data( - self, - request_id: str, - timeout: Optional[float] = None, - ) -> Union[ - types.LemurStringResponse, - types.LemurQuestionResponse, - ]: - response = api.lemur_get_response_data( - client=_client.Client.get_default().http_client, - request_id=request_id, - http_timeout=timeout, - ) - - return response - - -class Lemur: - """ - AssemblyAI's LeMUR (Leveraging Large Language Models to Understand Recognized Speech) framework - to process audio files with an LLM. - - See https://www.assemblyai.com/docs/Models/lemur for more information. - """ - - def __init__( - self, - sources: Optional[List[types.LemurSource]] = None, - client: Optional[_client.Client] = None, - ) -> None: - """ - Creates a new LeMUR instance to process audio files with an LLM. - - Args: - - sources: One or a list of sources to process (e.g. a `Transcript` or a `TranscriptGroup`) - client: The client to use for the LeMUR instance. If not provided, the default client will be used - """ - self._client = client or _client.Client.get_default() - - self._impl = _LemurImpl( - client=self._client, - sources=sources, - ) - self._executor = concurrent.futures.ThreadPoolExecutor() - - def question( - self, - questions: Union[types.LemurQuestion, List[types.LemurQuestion]], - context: Optional[Union[str, Dict[str, Any]]] = None, - final_model: Optional[types.LemurModel] = None, - max_output_size: Optional[int] = None, - timeout: Optional[float] = None, - temperature: Optional[float] = None, - input_text: Optional[str] = None, - ) -> types.LemurQuestionResponse: - """ - Question & Answer allows you to ask free form questions about one or many transcripts. - - This can be any question you find useful, such as judging the outcome or determining facts - about the audio. For instance, you can ask for action items from a meeting, did the customer - respond positively, or count how many times a word or phrase was said. - - See also Best Practices on LeMUR: https://www.assemblyai.com/docs/Guides/lemur_best_practices - - Args: - questions: One or a list of questions to ask. - context: The context which is shared among all questions. This can be a string or a dictionary. - final_model: The model that is used for the final prompt after compression is performed. - max_output_size: Max output size in tokens - timeout: The timeout in seconds to wait for the answer(s). - temperature: Change how deterministic the response is, with 0 being the most deterministic and 1 being the least deterministic. - input_text: Custom formatted transcript data. Use instead of transcript_ids. - - Returns: One or a list of answer objects. - """ - - if not isinstance(questions, list): - questions = [questions] - - return self._impl.question( - questions=questions, - context=context, - final_model=final_model, - max_output_size=max_output_size, - timeout=timeout, - temperature=temperature, - input_text=input_text, - ) - - def question_async( - self, - questions: Union[types.LemurQuestion, List[types.LemurQuestion]], - context: Optional[Union[str, Dict[str, Any]]] = None, - final_model: Optional[types.LemurModel] = None, - max_output_size: Optional[int] = None, - timeout: Optional[float] = None, - temperature: Optional[float] = None, - input_text: Optional[str] = None, - ) -> concurrent.futures.Future[types.LemurQuestionResponse]: - """ - Question & Answer allows you to ask free form questions about one or many transcripts. - - This can be any question you find useful, such as judging the outcome or determining facts - about the audio. For instance, you can ask for action items from a meeting, did the customer - respond positively, or count how many times a word or phrase was said. - - See also Best Practices on LeMUR: https://www.assemblyai.com/docs/Guides/lemur_best_practices - - Args: - questions: One or a list of questions to ask. - context: The context which is shared among all questions. This can be a string or a dictionary. - final_model: The model that is used for the final prompt after compression is performed. - max_output_size: Max output size in tokens - timeout: The timeout in seconds to wait for the answer(s). - temperature: Change how deterministic the response is, with 0 being the most deterministic and 1 being the least deterministic. - input_text: Custom formatted transcript data. Use instead of transcript_ids. - - Returns: One or a list of answer objects. - """ - - if not isinstance(questions, list): - questions = [questions] - - return self._executor.submit( - self._impl.question, - questions=questions, - context=context, - final_model=final_model, - max_output_size=max_output_size, - timeout=timeout, - temperature=temperature, - input_text=input_text, - ) - - def summarize( - self, - context: Optional[Union[str, Dict[str, Any]]] = None, - answer_format: Optional[str] = None, - final_model: Optional[types.LemurModel] = None, - max_output_size: Optional[int] = None, - timeout: Optional[float] = None, - temperature: Optional[float] = None, - input_text: Optional[str] = None, - ) -> types.LemurSummaryResponse: - """ - Summary allows you to distill a piece of audio into a few impactful sentences. - You can give the model context to get more pinpoint results while outputting the - results in a variety of formats described in human language. - - See also Best Practices on LeMUR: https://www.assemblyai.com/docs/Guides/lemur_best_practices - - Args: - context: An optional context on the transcript. - answer_format: The format on how the summary shall be summarized. - final_model: The model that is used for the final prompt after compression is performed. - max_output_size: Max output size in tokens - timeout: The timeout in seconds to wait for the summary. - temperature: Change how deterministic the response is, with 0 being the most deterministic and 1 being the least deterministic. - input_text: Custom formatted transcript data. Use instead of transcript_ids. - - Returns: The summary as a string. - """ - - return self._impl.summarize( - context=context, - answer_format=answer_format, - final_model=final_model, - max_output_size=max_output_size, - timeout=timeout, - temperature=temperature, - input_text=input_text, - ) - - def summarize_async( - self, - context: Optional[Union[str, Dict[str, Any]]] = None, - answer_format: Optional[str] = None, - final_model: Optional[types.LemurModel] = None, - max_output_size: Optional[int] = None, - timeout: Optional[float] = None, - temperature: Optional[float] = None, - input_text: Optional[str] = None, - ) -> concurrent.futures.Future[types.LemurSummaryResponse]: - """ - Summary allows you to distill a piece of audio into a few impactful sentences. - You can give the model context to get more pinpoint results while outputting the - results in a variety of formats described in human language. - - See also Best Practices on LeMUR: https://www.assemblyai.com/docs/Guides/lemur_best_practices - - Args: - context: An optional context on the transcript. - answer_format: The format on how the summary shall be summarized. - final_model: The model that is used for the final prompt after compression is performed. - max_output_size: Max output size in tokens - timeout: The timeout in seconds to wait for the summary. - temperature: Change how deterministic the response is, with 0 being the most deterministic and 1 being the least deterministic. - input_text: Custom formatted transcript data. Use instead of transcript_ids. - - Returns: The summary as a string. - """ - - return self._executor.submit( - self._impl.summarize, - context=context, - answer_format=answer_format, - final_model=final_model, - max_output_size=max_output_size, - timeout=timeout, - temperature=temperature, - input_text=input_text, - ) - - def action_items( - self, - context: Optional[Union[str, Dict[str, Any]]] = None, - answer_format: Optional[str] = None, - final_model: Optional[types.LemurModel] = None, - max_output_size: Optional[int] = None, - timeout: Optional[float] = None, - temperature: Optional[float] = None, - input_text: Optional[str] = None, - ) -> types.LemurActionItemsResponse: - """ - Action Items allows you to generate action items from one or many transcripts. - - You can provide the model with a context to get more pinpoint results while outputting the - results in a variety of formats described in human language. - - See also Best Practices on LeMUR: https://www.assemblyai.com/docs/Guides/lemur_best_practices - - Args: - context: An optional context on the transcript. - answer_format: The preferred format for the result action items. - final_model: The model that is used for the final prompt after compression is performed. - max_output_size: Max output size in tokens - timeout: The timeout in seconds to wait for the action items response. - temperature: Change how deterministic the response is, with 0 being the most deterministic and 1 being the least deterministic. - input_text: Custom formatted transcript data. Use instead of transcript_ids. - - Returns: The action items as a string. - """ - - return self._impl.action_items( - context=context, - answer_format=answer_format, - final_model=final_model, - max_output_size=max_output_size, - timeout=timeout, - temperature=temperature, - input_text=input_text, - ) - - def action_items_async( - self, - context: Optional[Union[str, Dict[str, Any]]] = None, - answer_format: Optional[str] = None, - final_model: Optional[types.LemurModel] = None, - max_output_size: Optional[int] = None, - timeout: Optional[float] = None, - temperature: Optional[float] = None, - input_text: Optional[str] = None, - ) -> concurrent.futures.Future[types.LemurActionItemsResponse]: - """ - Action Items allows you to generate action items from one or many transcripts. - - You can provide the model with a context to get more pinpoint results while outputting the - results in a variety of formats described in human language. - - See also Best Practices on LeMUR: https://www.assemblyai.com/docs/Guides/lemur_best_practices - - Args: - context: An optional context on the transcript. - answer_format: The preferred format for the result action items. - final_model: The model that is used for the final prompt after compression is performed. - max_output_size: Max output size in tokens - timeout: The timeout in seconds to wait for the action items response. - temperature: Change how deterministic the response is, with 0 being the most deterministic and 1 being the least deterministic. - input_text: Custom formatted transcript data. Use instead of transcript_ids. - - Returns: The action items as a string. - """ - - return self._executor.submit( - self._impl.action_items, - context=context, - answer_format=answer_format, - final_model=final_model, - max_output_size=max_output_size, - timeout=timeout, - temperature=temperature, - input_text=input_text, - ) - - def task( - self, - prompt: str, - context: Optional[Union[str, Dict[str, Any]]] = None, - final_model: Optional[types.LemurModel] = None, - max_output_size: Optional[int] = None, - timeout: Optional[float] = None, - temperature: Optional[float] = None, - input_text: Optional[str] = None, - ) -> types.LemurTaskResponse: - """ - Task feature allows you to submit a custom prompt to the model. - - See also Best Practices on LeMUR: https://www.assemblyai.com/docs/Guides/lemur_best_practices - - Args: - prompt: The prompt to use for this task. - context: An optional context on the transcript. - final_model: The model that is used for the final prompt after compression is performed. - max_output_size: Max output size in tokens - timeout: The timeout in seconds to wait for the task. - temperature: Change how deterministic the response is, with 0 being the most deterministic and 1 being the least deterministic. - input_text: Custom formatted transcript data. Use instead of transcript_ids. - - Returns: A response to a question or task submitted via custom prompt (with source transcripts or other sources taken into the context) - """ - - return self._impl.task( - prompt=prompt, - context=context, - final_model=final_model, - max_output_size=max_output_size, - timeout=timeout, - temperature=temperature, - input_text=input_text, - ) - - def task_async( - self, - prompt: str, - context: Optional[Union[str, Dict[str, Any]]] = None, - final_model: Optional[types.LemurModel] = None, - max_output_size: Optional[int] = None, - timeout: Optional[float] = None, - temperature: Optional[float] = None, - input_text: Optional[str] = None, - ) -> concurrent.futures.Future[types.LemurTaskResponse]: - """ - Task feature allows you to submit a custom prompt to the model. - - See also Best Practices on LeMUR: https://www.assemblyai.com/docs/Guides/lemur_best_practices - - Args: - prompt: The prompt to use for this task. - context: An optional context on the transcript. - final_model: The model that is used for the final prompt after compression is performed. - max_output_size: Max output size in tokens - timeout: The timeout in seconds to wait for the task. - temperature: Change how deterministic the response is, with 0 being the most deterministic and 1 being the least deterministic. - input_text: Custom formatted transcript data. Use instead of transcript_ids. - - Returns: A response to a question or task submitted via custom prompt (with source transcripts or other sources taken into the context) - """ - - return self._executor.submit( - self._impl.task, - prompt=prompt, - context=context, - final_model=final_model, - max_output_size=max_output_size, - timeout=timeout, - temperature=temperature, - input_text=input_text, - ) - - @classmethod - def purge_request_data( - cls, - request_id: str, - timeout: Optional[float] = None, - ) -> types.LemurPurgeResponse: - """ - Purge sent LeMUR request data that was previously sent. - - Args: - request_id: The request ID that was returned to you from the original LeMUR request that should be purged. - - Returns: A response saying whether the LeMUR request data was successfully purged. - """ - return _LemurImpl.purge_request_data( - request_id=request_id, - timeout=timeout, - ) - - @classmethod - def purge_request_data_async( - cls, - request_id: str, - timeout: Optional[float] = None, - ) -> concurrent.futures.Future[types.LemurPurgeResponse]: - """ - Purge sent LeMUR request data that was previously sent. - - Args: - request_id: The request ID that was returned to you from the original LeMUR request that should be purged. - - Returns: A response saying whether the LeMUR request data was successfully purged. - """ - with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: - response_future = executor.submit( - _LemurImpl.purge_request_data, - request_id=request_id, - timeout=timeout, - ) - return response_future - - def get_response_data( - self, - request_id: str, - timeout: Optional[float] = None, - ) -> Union[ - types.LemurStringResponse, - types.LemurQuestionResponse, - ]: - """ - Retrieve a LeMUR response that was previously generated. - - Args: - request_id: The ID of a previous LeMUR request. - timeout: The timeout in seconds to wait for the task. - - Returns: A LeMUR response that was previously generated. - """ - return self._impl.get_response_data(request_id=request_id, timeout=timeout) - - def get_response_data_async( - self, - request_id: str, - timeout: Optional[float] = None, - ) -> concurrent.futures.Future[ - Union[ - types.LemurStringResponse, - types.LemurQuestionResponse, - ] - ]: - """ - Retrieve a LeMUR response that was previously generated. - - Args: - request_id: The ID of a previous LeMUR request. - timeout: The timeout in seconds to wait for the task. - - Returns: A LeMUR response that was previously generated. - """ - return self._executor.submit( - self._impl.get_response_data, - request_id=request_id, - timeout=timeout, - ) diff --git a/assemblyai/prerecorded/v2/async_transcript.py b/assemblyai/prerecorded/v2/async_transcript.py index d32f873..587dc5a 100644 --- a/assemblyai/prerecorded/v2/async_transcript.py +++ b/assemblyai/prerecorded/v2/async_transcript.py @@ -30,7 +30,7 @@ def _open_binary(path: str, mode: str) -> BinaryIO: return cast(BinaryIO, open(path, mode)) -class AsyncTranscript(_BaseTranscript, types.Sourcable): +class AsyncTranscript(_BaseTranscript): """ The asyncio counterpart of `Transcript`. diff --git a/assemblyai/prerecorded/v2/transcript.py b/assemblyai/prerecorded/v2/transcript.py index 46b14c2..a23b8a2 100644 --- a/assemblyai/prerecorded/v2/transcript.py +++ b/assemblyai/prerecorded/v2/transcript.py @@ -11,7 +11,7 @@ from typing_extensions import Self from ... import client as _client -from ... import lemur, types +from ... import types from . import api from ._base import TERMINAL_STATUSES, _BaseTranscript, config_from_response @@ -197,7 +197,7 @@ def save_redacted_audio(self, filepath: str): f.write(chunk) @classmethod - def delete_by_id(cls, transcript_id: str) -> types.Transcript: + def delete_by_id(cls, transcript_id: str) -> Transcript: client = _client.Client.get_default() response = api.delete_transcript( client=client.http_client, transcript_id=transcript_id @@ -206,7 +206,7 @@ def delete_by_id(cls, transcript_id: str) -> types.Transcript: return Transcript.from_response(client=client, response=response) -class Transcript(_BaseTranscript, types.Sourcable): +class Transcript(_BaseTranscript): """ Transcript object to perform operations on the actual transcript. """ @@ -277,7 +277,7 @@ def get_by_id_async(cls, transcript_id: str) -> concurrent.futures.Future[Self]: return cls(transcript_id=transcript_id).wait_for_completion_async() @classmethod - def delete_by_id(cls, transcript_id: str) -> types.Transcript: + def delete_by_id(cls, transcript_id: str) -> Transcript: """Delete an existing transcript. Blocks until the transcript is completed. Args: @@ -291,7 +291,7 @@ def delete_by_id(cls, transcript_id: str) -> types.Transcript: @classmethod def delete_by_id_async( cls, transcript_id: str - ) -> concurrent.futures.Future[types.Transcript]: + ) -> concurrent.futures.Future[Transcript]: """Delete an existing transcript asynchronously. Args: @@ -319,17 +319,6 @@ def _response(self) -> types.TranscriptResponse: return self._impl.transcript - @property - def lemur(self) -> lemur.Lemur: - """ - Access AssemblyAI's LeMUR features. - """ - - return lemur.Lemur( - client=self._client, - sources=[types.LemurSource(self)], - ) - def export_subtitles_srt( self, chars_per_caption: Optional[int] = None, diff --git a/assemblyai/prerecorded/v2/transcript_group.py b/assemblyai/prerecorded/v2/transcript_group.py index e44f3e7..ef64629 100644 --- a/assemblyai/prerecorded/v2/transcript_group.py +++ b/assemblyai/prerecorded/v2/transcript_group.py @@ -8,7 +8,7 @@ from typing_extensions import Self from ... import client as _client -from ... import lemur, types +from ... import types from .transcript import Transcript @@ -143,17 +143,6 @@ def status(self) -> types.TranscriptStatus: else: raise ValueError(f"Unexpected status type: {all_status}") - @property - def lemur(self) -> lemur.Lemur: - """ - Access AssemblyAI's LeMUR functionality. - """ - - return lemur.Lemur( - client=self._impl._client, - sources=[types.LemurSource(t) for t in self.transcripts], - ) - def add_transcript( self, transcript: Union[Transcript, str], diff --git a/assemblyai/types.py b/assemblyai/types.py index 3dc2ec3..01ad4f4 100644 --- a/assemblyai/types.py +++ b/assemblyai/types.py @@ -1,8 +1,6 @@ import sys -from datetime import datetime from enum import Enum, EnumMeta from typing import ( - TYPE_CHECKING, Annotated, Any, Dict, @@ -16,9 +14,6 @@ from urllib.parse import parse_qs, urlparse from warnings import warn -if TYPE_CHECKING: - from .prerecorded.v2.transcript import Transcript - try: # pydantic v2 import from pydantic import BaseModel, ConfigDict, Field, field_validator @@ -78,12 +73,6 @@ class RedactedAudioUnavailableError(AssemblyAIError): """ -class LemurError(AssemblyAIError): - """ - Error class when a Lemur request fails - """ - - class SyncTranscriptError(AssemblyAIError): """ Error raised when a synchronous transcription request fails. @@ -107,14 +96,6 @@ def __init__( self.retry_after = retry_after -class Sourcable: - """ - A base class for all sourcable objects - - Currently, only `Transcript` is sourcable - """ - - class Settings(BaseSettings): """ Settings for the AssemblyAI client @@ -2658,344 +2639,6 @@ class ListTranscriptResponse(BaseModel): "A list of transcripts sorted from newest to oldest" -class LemurSourceType(str, Enum): - """ - The source type of the LeMUR request - """ - - transcript = "transcript" - "The source is a transcript" - - -class LemurSource: - """ - A LeMUR source is a source that can be used to process it with an LLM. - """ - - def __init__( - self, - source: Sourcable, - ) -> None: - """ - Creates a new LeMUR source to process audio files with an LLM. - - Args: - - source: The source to process (e.g. a `Transcript`) - """ - self._source = source - self._type = None - - from . import AsyncTranscript, Transcript - - # LeMUR is sync-only for now, but only a source's id travels to the - # API, so an `AsyncTranscript` works just as well. - if isinstance(source, (Transcript, AsyncTranscript)): - self._type = LemurSourceType.transcript - else: - raise ValueError(f"Invalid source: {source}") - - @property - def source(self) -> Sourcable: - """ - The source to process (e.g. a `Transcript`) - """ - return self._source - - @property - def type(self) -> LemurSourceType: - """ - The type of the source. - """ - return self._type # type: ignore - - -class LemurTranscriptSource(LemurSource): - """ - A LeMUR source that can be used to process a transcript with an LLM. - """ - - def __init__( - self, - transcript: Union["Transcript", str], - ) -> None: - """ - Creates a new LeMUR transcript source to process audio files with an LLM. - - Args: - - transcript: The transcript to process - context: An optional context on the source (can be a string or an arbitrary dictionary) - """ - from . import Transcript - - if isinstance(transcript, str): - transcript = Transcript(transcript_id=transcript) - - super().__init__(transcript) - - -class LemurSourceRequest(BaseModel): - id: Optional[str] = None - "The unique identifier of your source - only relevant for transcript sources" - - type: LemurSourceType - "The type of source" - - @classmethod - def from_lemur_source(cls, source: LemurSource) -> Self: - """ - Creates a LemurSourceRequest from a LemurSource - """ - if source.type == LemurSourceType.transcript: - return cls( - id=source.source.id, # type:ignore - type=source.type, - ) - - raise ValueError("Unsupported source type") - - -class LemurModel(str, Enum): - """ - LeMUR features different model modes that allow you to configure your request to suit your needs. - """ - - claude_opus_4_20250514 = "anthropic/claude-opus-4-20250514" - """ - https://www.anthropic.com/news/claude-4 - read to understand the capabilities/limitations of opus 4 - - Input Token Limit: 200k - Output Token Limit: 64k - """ - - claude_sonnet_4_20250514 = "anthropic/claude-sonnet-4-20250514" - """ - https://www.anthropic.com/news/claude-4 - read to understand the capabilities/limitations of sonnet 4 - - Input Token Limit: 200k - Output Token Limit: 64k - """ - - claude3_7_sonnet_20250219 = "anthropic/claude-3-7-sonnet-20250219" - """ - https://www.anthropic.com/news/claude-3-7-sonnet - read to understand the capabilities/limitations of sonnet 3.7 - """ - - claude3_5_sonnet = "anthropic/claude-3-5-sonnet" - """ - https://www.anthropic.com/news/claude-3-5-sonnet - read to understand the capabilities/limitations of sonnet 3.7 - """ - - claude3_opus = "anthropic/claude-3-opus" - """ - Deprecated: EOL ~Jan 2026 - """ - - claude3_5_haiku_20241022 = "anthropic/claude-3-5-haiku-20241022" - """ - https://www.anthropic.com/claude/haiku - read to understand the capabilities/limitations of haiku 3.5 - """ - - claude3_haiku = "anthropic/claude-3-haiku" - """ - https://www.anthropic.com/news/claude-3-haiku - read to understand the capabilities/limitations of haiku 3 - """ - - claude3_sonnet = "anthropic/claude-3-sonnet" - """ - Deprecated: EOL ~July 2025 - Claude 3 Sonnet is a legacy model with a balanced combination of performance and speed for efficient, high-throughput tasks. - """ - - claude2_1 = "anthropic/claude-2-1" - """ - Deprecated: Claude 2.1 is deprecated and will stop working on Feb 6th, 2025. - """ - - claude2_0 = "anthropic/claude-2" - """ - Deprecated: Claude 2.0 is deprecated and will stop working on Feb 6th, 2025. - """ - - default = "default" - """ - Deprecated: Legacy model. The same as `claude2_0` and will stop working on Feb 6th, 2025. - """ - - mistral7b = "assemblyai/mistral-7b" - """ - Mistral 7B is an open source model that works well for summarization and answering questions. - """ - - -class LemurQuestionAnswer(BaseModel): - """ - The result of your Question and Answer LeMUR request. - """ - - question: str - "The question that was asked" - - answer: str - "The answer to the question" - - -class LemurQuestion(BaseModel): - """ - The question you wish to ask LeMUR - """ - - question: str - "The question you wish to ask" - - context: Optional[Union[str, Dict[str, Any]]] = None - "Context to provide the model - this can be a string or an arbitrary dictionary" - - answer_format: Optional[str] = None - """ - How you want the answer to be returned. This can be any text. - Cannot be used with answer_options. - - Examples: - - - "short sentence" - - "bullet points" - """ - - answer_options: Optional[List[str]] = None - """ - What discrete options to return. Useful for precise responses. - - Cannot be used with answer_format. - - Examples: - - - ["Yes", "No"] - - ["High", "Medium", "Low"] - """ - - -class BaseLemurRequest(BaseModel): - sources: List[LemurSourceRequest] - final_model: Optional[LemurModel] = None - max_output_size: Optional[int] = None - temperature: Optional[float] = None - input_text: Optional[str] = None - - -class LemurUsage(BaseModel): - """ - The usage numbers for the LeMUR request - """ - - input_tokens: int - "The number of input tokens used by the model" - - output_tokens: int - "The number of output tokens generated by the model" - - -class LemurRequestDetails(BaseModel): - request_endpoint: str - temperature: float - final_model: str - max_output_size: int - created_at: datetime - transcript_ids: Optional[List[str]] = None - input_text: Optional[str] = None - questions: Optional[List[LemurQuestion]] = None - prompt: Optional[str] = None - context: Optional[Union[dict, str]] = None - answer_format: Optional[str] = None - - -class BaseLemurResponse(BaseModel): - request_id: str - "The unique identifier of your LeMUR request" - - usage: LemurUsage - "The usage numbers for the LeMUR request" - - request: Optional[LemurRequestDetails] = None - "The request details the user passed into the POST request. Optional since this only exists on the GET request." - - -class LemurStringResponse(BaseLemurResponse): - """ - The result of your LeMUR request with a string response. - """ - - response: str - "The LLM response to your request" - - -class LemurTaskRequest(BaseLemurRequest): - context: Optional[Union[str, Dict[str, Any]]] = None - prompt: str - - -class LemurTaskResponse(LemurStringResponse): - """ - The result of your LeMUR Task request. - """ - - -class LemurQuestionRequest(BaseLemurRequest): - context: Optional[Union[str, Dict[str, Any]]] = None - questions: List[LemurQuestion] - - -class LemurQuestionResponse(BaseLemurResponse): - """ - The result of your Question and Answer LeMUR request. - """ - - response: List[LemurQuestionAnswer] - "The list of answers to your questions" - - -class LemurSummaryRequest(BaseLemurRequest): - context: Optional[Union[str, Dict[str, Any]]] = None - answer_format: Optional[str] = None - - -class LemurSummaryResponse(LemurStringResponse): - """ - The result of your Summary LeMUR request. - """ - - -class LemurActionItemsRequest(BaseLemurRequest): - context: Optional[Union[str, Dict[str, Any]]] = None - answer_format: Optional[str] = None - - -class LemurActionItemsResponse(LemurStringResponse): - """ - The result of your Action Items LeMUR request. - """ - - -class LemurPurgeRequest(BaseModel): - request_id: str - - -class LemurPurgeResponse(BaseModel): - """ - The result of your LeMUR purge request. - """ - - request_id: str - "The unique identifier of the LeMUR purge request" - - request_id_to_purge: str - "The unique identifier of the LeMUR request nneds to be purged" - - deleted: bool - "The result of the LeMUR purge request" - - # Caps mirror the sync service's `config` part. `prompt` and `keyterms_prompt` # over their caps are rejected; `conversation_context` over its caps is # trimmed (oldest turns first), matching the server. diff --git a/tests/unit/factories.py b/tests/unit/factories.py index ef684ac..7288c8a 100644 --- a/tests/unit/factories.py +++ b/tests/unit/factories.py @@ -246,161 +246,6 @@ class Meta: ) -class LemurRequestDetails(factory.Factory): - class Meta: - model = types.LemurRequestDetails - - request_endpoint = factory.Faker("text") - temperature = factory.Faker("pyfloat") - final_model = factory.Faker("text") - max_output_size = factory.Faker("pyint") - created_at = factory.Faker("iso8601") - - -class LemurTaskRequestDetails(LemurRequestDetails): - """Request details specific to LeMUR task operations""" - - request_endpoint = "/lemur/v3/task" - prompt = factory.Faker("text") - - -class LemurSummaryRequestDetails(LemurRequestDetails): - """Request details specific to LeMUR summary operations""" - - request_endpoint = "/lemur/v3/summary" - context = factory.LazyFunction(lambda: {"key": "value"}) - answer_format = factory.Faker("sentence") - - -class LemurQuestionRequestDetails(LemurRequestDetails): - """Request details specific to LeMUR question-answer operations""" - - request_endpoint = "/lemur/v3/question-answer" - questions = [ - { - "question": "What is the main topic?", - "answer_format": "short sentence", - "context": "Meeting context", - }, - { - "question": "What is the sentiment?", - "answer_options": ["positive", "negative", "neutral"], - }, - ] - - -class LemurUsage(factory.Factory): - class Meta: - model = types.LemurUsage - - input_tokens = factory.Faker("pyint") - output_tokens = factory.Faker("pyint") - - -class LemurQuestionAnswer(factory.Factory): - class Meta: - model = types.LemurQuestionAnswer - - question = factory.Faker("text") - answer = factory.Faker("text") - - -class LemurQuestionResponse(factory.Factory): - class Meta: - model = types.LemurQuestionResponse - - request_id = factory.Faker("uuid4") - usage = factory.SubFactory(LemurUsage) - response = factory.List( - [ - factory.SubFactory(LemurQuestionAnswer), - factory.SubFactory(LemurQuestionAnswer), - ] - ) - - -class LemurSummaryResponse(factory.Factory): - class Meta: - model = types.LemurSummaryResponse - - request_id = factory.Faker("uuid4") - usage = factory.SubFactory(LemurUsage) - response = factory.Faker("text") - - -class LemurActionItemsResponse(factory.Factory): - class Meta: - model = types.LemurActionItemsResponse - - request_id = factory.Faker("uuid4") - usage = factory.SubFactory(LemurUsage) - response = factory.Faker("text") - - -class LemurTaskResponse(factory.Factory): - class Meta: - model = types.LemurTaskResponse - - request_id = factory.Faker("uuid4") - usage = factory.SubFactory(LemurUsage) - response = factory.Faker("text") - - -class LemurStringResponse(factory.Factory): - class Meta: - model = types.LemurStringResponse - - request_id = factory.Faker("uuid4") - usage = factory.SubFactory(LemurUsage) - response = factory.Faker("text") - request = factory.SubFactory(LemurRequestDetails) - - -# Factories specifically for get_response endpoint tests (include request field) -class LemurTaskResponseWithRequest(factory.Factory): - class Meta: - model = types.LemurTaskResponse - - request_id = factory.Faker("uuid4") - usage = factory.SubFactory(LemurUsage) - response = factory.Faker("text") - request = factory.SubFactory(LemurTaskRequestDetails) - - -class LemurSummaryResponseWithRequest(factory.Factory): - class Meta: - model = types.LemurSummaryResponse - - request_id = factory.Faker("uuid4") - usage = factory.SubFactory(LemurUsage) - response = factory.Faker("text") - request = factory.SubFactory(LemurSummaryRequestDetails) - - -class LemurQuestionResponseWithRequest(factory.Factory): - class Meta: - model = types.LemurQuestionResponse - - request_id = factory.Faker("uuid4") - usage = factory.SubFactory(LemurUsage) - response = factory.List( - [ - factory.SubFactory(LemurQuestionAnswer), - factory.SubFactory(LemurQuestionAnswer), - ] - ) - request = factory.SubFactory(LemurQuestionRequestDetails) - - -class LemurPurgeResponse(factory.Factory): - class Meta: - model = types.LemurPurgeResponse - - request_id = factory.Faker("uuid4") - request_id_to_purge = factory.Faker("uuid4") - deleted = True - - class WordSearchMatchFactory(factory.Factory): class Meta: model = types.WordSearchMatch diff --git a/tests/unit/test_async_transcriber.py b/tests/unit/test_async_transcriber.py index d141650..bdcd683 100644 --- a/tests/unit/test_async_transcriber.py +++ b/tests/unit/test_async_transcriber.py @@ -901,20 +901,6 @@ async def test_client_requires_an_api_key(): aai.AsyncClient(settings=settings) -async def test_async_transcript_is_a_lemur_source(httpx_mock: HTTPXMock): - # Given a completed async transcript - completed = _completed_response() - _mock_submit(httpx_mock, completed) - - async with aai.AsyncTranscriber() as transcriber: - transcript = await transcriber.submit("https://example.org/audio.wav") - - # Then it can be handed to LeMUR, which only needs its id - source = aai.LemurSource(transcript) - - assert source.source.id == completed["id"] - - async def test_both_concurrency_models_share_one_base(): """The asyncio classes inherit the same bases as the threaded ones.""" assert issubclass(aai.AsyncTranscript, _BaseTranscript) diff --git a/tests/unit/test_lemur.py b/tests/unit/test_lemur.py deleted file mode 100644 index 4696a98..0000000 --- a/tests/unit/test_lemur.py +++ /dev/null @@ -1,1168 +0,0 @@ -import uuid - -import httpx -import pytest -from pytest_httpx import HTTPXMock - -import assemblyai as aai -from assemblyai.api import ( - ENDPOINT_LEMUR, - ENDPOINT_LEMUR_BASE, -) -from tests.unit import factories - -aai.settings.api_key = "test" - - -def test_lemur_single_question_succeeds_transcript(httpx_mock: HTTPXMock): - """ - Tests whether asking a single question succeeds. - """ - - # create a mock response of a LemurQuestionResponse - mock_lemur_answer = factories.generate_dict_factory( - factories.LemurQuestionResponse - )() - - # we only want to mock one answer - mock_lemur_answer["response"] = [mock_lemur_answer["response"][0]] - - # prepare the question to be asked - question = aai.LemurQuestion( - question="Which cars do the callers want to buy?", - context="Callers are interested in buying cars", - answer_options=["Toyota", "Honda", "Ford", "Chevrolet"], - ) - - # update the mock question with the question - mock_lemur_answer["response"][0]["question"] = question.question - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR}/question-answer", - status_code=httpx.codes.OK, - method="POST", - json=mock_lemur_answer, - ) - - transcript = aai.Transcript(str(uuid.uuid4())) - - # mimic the usage of the SDK - lemur = aai.Lemur(sources=[aai.LemurSource(transcript)]) - result = lemur.question(question) - - # check whether answer is not a list - assert isinstance(result, aai.LemurQuestionResponse) - - answers = result.response - - # check the response - assert answers[0].question == mock_lemur_answer["response"][0]["question"] - assert answers[0].answer == mock_lemur_answer["response"][0]["answer"] - - assert result.usage.input_tokens == mock_lemur_answer["usage"]["input_tokens"] - assert result.usage.output_tokens == mock_lemur_answer["usage"]["output_tokens"] - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_single_question_succeeds_input_text(httpx_mock: HTTPXMock): - """ - Tests whether asking a single question succeeds with input text. - """ - - # create a mock response of a LemurQuestionResponse - mock_lemur_answer = factories.generate_dict_factory( - factories.LemurQuestionResponse - )() - - # we only want to mock one answer - mock_lemur_answer["response"] = [mock_lemur_answer["response"][0]] - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR}/question-answer", - status_code=httpx.codes.OK, - method="POST", - json=mock_lemur_answer, - ) - - # prepare the question to be asked - question = aai.LemurQuestion( - question="Which cars do the callers want to buy?", - context="Callers are interested in buying cars", - answer_options=["Toyota", "Honda", "Ford", "Chevrolet"], - ) - # test input_text input - # mimic the usage of the SDK - lemur = aai.Lemur() - result = lemur.question( - question, input_text="This transcript is a test transcript." - ) - - # check whether answer is not a list - assert isinstance(result, aai.LemurQuestionResponse) - - answers = result.response - - # check the response - assert answers[0].question == mock_lemur_answer["response"][0]["question"] - assert answers[0].answer == mock_lemur_answer["response"][0]["answer"] - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_multiple_question_succeeds_transcript(httpx_mock: HTTPXMock): - """ - Tests whether asking multiple questions succeeds. - """ - - # create a mock response of a LemurQuestionResponse - mock_lemur_answer = factories.generate_dict_factory( - factories.LemurQuestionResponse - )() - - # prepare the questions to be asked - questions = [ - aai.LemurQuestion( - question="Which cars do the callers want to buy?", - ), - aai.LemurQuestion( - question="What price range are the callers looking for?", - ), - ] - - # update the mock questions with the questions - mock_lemur_answer["response"][0]["question"] = questions[0].question - mock_lemur_answer["response"][1]["question"] = questions[1].question - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR}/question-answer", - status_code=httpx.codes.OK, - method="POST", - json=mock_lemur_answer, - ) - - transcript = aai.Transcript(str(uuid.uuid4())) - - # mimic the usage of the SDK - lemur = aai.Lemur(sources=[aai.LemurSource(transcript)]) - result = lemur.question(questions=questions) - - assert isinstance(result, aai.LemurQuestionResponse) - - answers = result.response - # check whether answers is a list - assert isinstance(answers, list) - - # check the response - for idx, answer in enumerate(answers): - assert answer.question == mock_lemur_answer["response"][idx]["question"] - assert answer.answer == mock_lemur_answer["response"][idx]["answer"] - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_multiple_question_succeeds_input_text(httpx_mock: HTTPXMock): - """ - Tests whether asking multiple questions succeeds. - """ - - # create a mock response of a LemurQuestionResponse - mock_lemur_answer = factories.generate_dict_factory( - factories.LemurQuestionResponse - )() - - # prepare the questions to be asked - questions = [ - aai.LemurQuestion( - question="Which cars do the callers want to buy?", - ), - aai.LemurQuestion( - question="What price range are the callers looking for?", - ), - ] - - # update the mock questions with the questions - mock_lemur_answer["response"][0]["question"] = questions[0].question - mock_lemur_answer["response"][1]["question"] = questions[1].question - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR}/question-answer", - status_code=httpx.codes.OK, - method="POST", - json=mock_lemur_answer, - ) - - # test input_text input - # mimic the usage of the SDK - lemur = aai.Lemur() - result = lemur.question( - questions, input_text="This transcript is a test transcript." - ) - assert isinstance(result, aai.LemurQuestionResponse) - - answers = result.response - # check whether answers is a list - assert isinstance(answers, list) - - # check the response - for idx, answer in enumerate(answers): - assert answer.question == mock_lemur_answer["response"][idx]["question"] - assert answer.answer == mock_lemur_answer["response"][idx]["answer"] - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_question_fails(httpx_mock: HTTPXMock): - """ - Tests whether asking a question fails. - """ - - # prepare the question to be asked - question = aai.LemurQuestion( - question="Which cars do the callers want to buy?", - context="Callers are interested in buying cars", - answer_options=["Toyota", "Honda", "Ford", "Chevrolet"], - ) - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR}/question-answer", - status_code=httpx.codes.INTERNAL_SERVER_ERROR, - method="POST", - json={"error": "something went wrong"}, - ) - - # mimic the usage of the SDK - transcript = aai.Transcript(str(uuid.uuid4())) - - lemur = aai.Lemur(sources=[aai.LemurSource(transcript)]) - - with pytest.raises(aai.LemurError): - lemur.question(question) - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_summarize_succeeds_transcript(httpx_mock: HTTPXMock): - """ - Tests whether summarizing a transcript via LeMUR succeeds. - """ - - # create a mock response of a LemurSummaryResponse - mock_lemur_summary = factories.generate_dict_factory( - factories.LemurSummaryResponse - )() - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR}/summary", - status_code=httpx.codes.OK, - method="POST", - json=mock_lemur_summary, - ) - - # mimic the usage of the SDK - transcript = aai.Transcript(str(uuid.uuid4())) - - lemur = aai.Lemur(sources=[aai.LemurSource(transcript)]) - result = lemur.summarize(context="Callers asking for cars", answer_format="TLDR") - - assert isinstance(result, aai.LemurSummaryResponse) - - summary = result.response - - # check the response - assert summary == mock_lemur_summary["response"] - - assert result.usage.input_tokens == mock_lemur_summary["usage"]["input_tokens"] - assert result.usage.output_tokens == mock_lemur_summary["usage"]["output_tokens"] - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_summarize_succeeds_input_text(httpx_mock: HTTPXMock): - """ - Tests whether summarizing a transcript via LeMUR succeeds with input text. - """ - - # create a mock response of a LemurSummaryResponse - mock_lemur_summary = factories.generate_dict_factory( - factories.LemurSummaryResponse - )() - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR}/summary", - status_code=httpx.codes.OK, - method="POST", - json=mock_lemur_summary, - ) - - # test input_text input - lemur = aai.Lemur() - result = lemur.summarize( - context="Callers asking for cars", answer_format="TLDR", input_text="Test test" - ) - - assert isinstance(result, aai.LemurSummaryResponse) - - summary = result.response - - # check the response - assert summary == mock_lemur_summary["response"] - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_summarize_fails(httpx_mock: HTTPXMock): - """ - Tests whether summarizing a transcript via LeMUR fails. - """ - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR}/summary", - status_code=httpx.codes.INTERNAL_SERVER_ERROR, - method="POST", - json={"error": "something went wrong"}, - ) - - # mimic the usage of the SDK - transcript = aai.Transcript(str(uuid.uuid4())) - - lemur = aai.Lemur(sources=[aai.LemurSource(transcript)]) - - with pytest.raises(aai.LemurError): - lemur.summarize(context="Callers asking for cars", answer_format="TLDR") - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_action_items_succeeds_transcript(httpx_mock: HTTPXMock): - """ - Tests whether generating action items for a transcript via LeMUR succeeds. - """ - - # create a mock response of a LemurActionItemsResponse - mock_lemur_action_items = factories.generate_dict_factory( - factories.LemurActionItemsResponse - )() - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR}/action-items", - status_code=httpx.codes.OK, - method="POST", - json=mock_lemur_action_items, - ) - - # mimic the usage of the SDK - transcript = aai.Transcript(str(uuid.uuid4())) - - lemur = aai.Lemur(sources=[aai.LemurSource(transcript)]) - result = lemur.action_items( - context="Customers asking for help with resolving their problem", - answer_format="Three bullet points", - ) - - assert isinstance(result, aai.LemurActionItemsResponse) - - action_items = result.response - - # check the response - assert action_items == mock_lemur_action_items["response"] - - assert result.usage.input_tokens == mock_lemur_action_items["usage"]["input_tokens"] - assert ( - result.usage.output_tokens == mock_lemur_action_items["usage"]["output_tokens"] - ) - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_action_items_succeeds_input_text(httpx_mock: HTTPXMock): - """ - Tests whether generating action items for a transcript via LeMUR succeeds. - """ - - # create a mock response of a LemurActionItemsResponse - mock_lemur_action_items = factories.generate_dict_factory( - factories.LemurActionItemsResponse - )() - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR}/action-items", - status_code=httpx.codes.OK, - method="POST", - json=mock_lemur_action_items, - ) - - # test input_text input - lemur = aai.Lemur() - result = lemur.action_items( - context="Customers asking for help with resolving their problem", - answer_format="Three bullet points", - input_text="Test test", - ) - - assert isinstance(result, aai.LemurActionItemsResponse) - - action_items = result.response - - # check the response - assert action_items == mock_lemur_action_items["response"] - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_action_items_fails(httpx_mock: HTTPXMock): - """ - Tests whether generating action items for a transcript via LeMUR fails. - """ - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR}/action-items", - status_code=httpx.codes.INTERNAL_SERVER_ERROR, - method="POST", - json={"error": "something went wrong"}, - ) - - # mimic the usage of the SDK - transcript = aai.Transcript(str(uuid.uuid4())) - - lemur = aai.Lemur(sources=[aai.LemurSource(transcript)]) - - with pytest.raises(aai.LemurError): - lemur.action_items( - context="Customers asking for help with resolving their problem", - answer_format="Three bullet points", - ) - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_task_succeeds_transcript(httpx_mock: HTTPXMock): - """ - Tests whether creating a task request succeeds. - """ - - # create a mock response of a LemurTaskResponse - mock_lemur_task_response = factories.generate_dict_factory( - factories.LemurTaskResponse - )() - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR}/task", - status_code=httpx.codes.OK, - method="POST", - json=mock_lemur_task_response, - ) - - # mimic the usage of the SDK - transcript = aai.Transcript(str(uuid.uuid4())) - - lemur = aai.Lemur( - sources=[aai.LemurSource(transcript)], - ) - result = lemur.task( - prompt="Create action items of the meeting", context="An important meeting" - ) - - # check the response - assert isinstance(result, aai.LemurTaskResponse) - - assert result.response == mock_lemur_task_response["response"] - - assert ( - result.usage.input_tokens == mock_lemur_task_response["usage"]["input_tokens"] - ) - assert ( - result.usage.output_tokens == mock_lemur_task_response["usage"]["output_tokens"] - ) - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_task_succeeds_input_text(httpx_mock: HTTPXMock): - """ - Tests whether creating a task request succeeds. - """ - - # create a mock response of a LemurTaskResponse - mock_lemur_task_response = factories.generate_dict_factory( - factories.LemurTaskResponse - )() - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR}/task", - status_code=httpx.codes.OK, - method="POST", - json=mock_lemur_task_response, - ) - # test input_text input - lemur = aai.Lemur() - result = lemur.task( - prompt="Create action items of the meeting", input_text="Test test" - ) - - # check the response - assert isinstance(result, aai.LemurTaskResponse) - - assert result.response == mock_lemur_task_response["response"] - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -@pytest.mark.parametrize( - "final_model", - ( - aai.LemurModel.claude3_5_sonnet, - aai.LemurModel.claude3_opus, - aai.LemurModel.claude3_haiku, - aai.LemurModel.claude3_sonnet, - aai.LemurModel.claude2_1, - aai.LemurModel.claude2_0, - aai.LemurModel.default, - aai.LemurModel.mistral7b, - ), -) -def test_lemur_task_succeeds(final_model, httpx_mock: HTTPXMock): - """ - Tests whether creating a task request succeeds with other models. - """ - - # create a mock response of a LemurTaskResponse - mock_lemur_task_response = factories.generate_dict_factory( - factories.LemurTaskResponse - )() - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR}/task", - status_code=httpx.codes.OK, - method="POST", - json=mock_lemur_task_response, - ) - # test input_text input - lemur = aai.Lemur() - result = lemur.task( - final_model=final_model, - prompt="Create action items of the meeting", - context="An important meeting", - input_text="Test test", - ) - - # check the response - assert isinstance(result, aai.LemurTaskResponse) - - assert result.response == mock_lemur_task_response["response"] - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_ask_coach_fails(httpx_mock: HTTPXMock): - """ - Tests whether creating a task request fails. - """ - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR}/task", - status_code=httpx.codes.INTERNAL_SERVER_ERROR, - method="POST", - json={"error": "something went wrong"}, - ) - - # mimic the usage of the SDK - transcript = aai.Transcript(str(uuid.uuid4())) - - lemur = aai.Lemur(sources=[aai.LemurSource(transcript)]) - - with pytest.raises(aai.LemurError): - lemur.task(prompt="Create action items of the meeting") - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_purge_request_data_succeeds(httpx_mock: HTTPXMock): - """ - Tests whether LeMUR request purging succeeds. - """ - - # create a mock response of a LemurPurgeResponse - mock_lemur_purge_response = factories.generate_dict_factory( - factories.LemurPurgeResponse - )() - - mock_request_id: str = str(uuid.uuid4()) - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR_BASE}/{mock_request_id}", - status_code=httpx.codes.OK, - method="DELETE", - json=mock_lemur_purge_response, - ) - - # mimic the usage of the SDK - result = aai.Lemur.purge_request_data(request_id=mock_request_id) - - # check the response - assert isinstance(result, aai.LemurPurgeResponse) - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_purge_request_data_fails(httpx_mock: HTTPXMock): - """ - Tests whether LeMUR request purging fails. - """ - - # create a mock response of a LemurPurgeResponse - mock_lemur_purge_response = factories.generate_dict_factory( - factories.LemurPurgeResponse - )() - - mock_request_id: str = str(uuid.uuid4()) - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR_BASE}/{mock_request_id}", - status_code=httpx.codes.INTERNAL_SERVER_ERROR, - method="DELETE", - json=mock_lemur_purge_response, - ) - - with pytest.raises(aai.LemurError): - aai.Lemur.purge_request_data(mock_request_id) - - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_single_question_async_succeeds_transcript(httpx_mock: HTTPXMock): - """ - Tests whether asking a single question succeeds when async is used. - """ - - # create a mock response of a LemurQuestionResponse - mock_lemur_answer = factories.generate_dict_factory( - factories.LemurQuestionResponse - )() - - # we only want to mock one answer - mock_lemur_answer["response"] = [mock_lemur_answer["response"][0]] - - # prepare the question to be asked - question = aai.LemurQuestion( - question="Which cars do the callers want to buy?", - context="Callers are interested in buying cars", - answer_options=["Toyota", "Honda", "Ford", "Chevrolet"], - ) - - # update the mock question with the question - mock_lemur_answer["response"][0]["question"] = question.question - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR}/question-answer", - status_code=httpx.codes.OK, - method="POST", - json=mock_lemur_answer, - ) - - transcript = aai.Transcript(str(uuid.uuid4())) - - # mimic the usage of the SDK - lemur = aai.Lemur(sources=[aai.LemurSource(transcript)]) - result_future = lemur.question_async(question) - - result = result_future.result() - - # check whether answer is not a list - assert isinstance(result, aai.LemurQuestionResponse) - - answers = result.response - - # check the response - assert answers[0].question == mock_lemur_answer["response"][0]["question"] - assert answers[0].answer == mock_lemur_answer["response"][0]["answer"] - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_summarize_async_succeeds_transcript(httpx_mock: HTTPXMock): - """ - Tests whether summarizing a transcript via LeMUR succeeds - when async is used. - """ - - # create a mock response of a LemurSummaryResponse - mock_lemur_summary = factories.generate_dict_factory( - factories.LemurSummaryResponse - )() - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR}/summary", - status_code=httpx.codes.OK, - method="POST", - json=mock_lemur_summary, - ) - - # mimic the usage of the SDK - transcript = aai.Transcript(str(uuid.uuid4())) - - lemur = aai.Lemur(sources=[aai.LemurSource(transcript)]) - result_future = lemur.summarize_async( - context="Callers asking for cars", answer_format="TLDR" - ) - - result = result_future.result() - - assert isinstance(result, aai.LemurSummaryResponse) - - summary = result.response - - # check the response - assert summary == mock_lemur_summary["response"] - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_action_items_async_succeeds_transcript(httpx_mock: HTTPXMock): - """ - Tests whether generating action items for a transcript via LeMUR succeeds - when async is used. - """ - - # create a mock response of a LemurActionItemsResponse - mock_lemur_action_items = factories.generate_dict_factory( - factories.LemurActionItemsResponse - )() - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR}/action-items", - status_code=httpx.codes.OK, - method="POST", - json=mock_lemur_action_items, - ) - - # mimic the usage of the SDK - transcript = aai.Transcript(str(uuid.uuid4())) - - lemur = aai.Lemur(sources=[aai.LemurSource(transcript)]) - result_future = lemur.action_items_async( - context="Customers asking for help with resolving their problem", - answer_format="Three bullet points", - ) - - result = result_future.result() - - assert isinstance(result, aai.LemurActionItemsResponse) - - action_items = result.response - - # check the response - assert action_items == mock_lemur_action_items["response"] - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_task_async_succeeds_transcript(httpx_mock: HTTPXMock): - """ - Tests whether creating a task request succeeds when async is used. - """ - - # create a mock response of a LemurTaskResponse - mock_lemur_task_response = factories.generate_dict_factory( - factories.LemurTaskResponse - )() - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR}/task", - status_code=httpx.codes.OK, - method="POST", - json=mock_lemur_task_response, - ) - - # mimic the usage of the SDK - transcript = aai.Transcript(str(uuid.uuid4())) - - lemur = aai.Lemur( - sources=[aai.LemurSource(transcript)], - ) - result_future = lemur.task_async(prompt="Create action items of the meeting") - - result = result_future.result() - - # check the response - assert isinstance(result, aai.LemurTaskResponse) - - assert result.response == mock_lemur_task_response["response"] - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_purge_request_data_async_succeeds(httpx_mock: HTTPXMock): - """ - Tests whether LeMUR request purging succeeds when async is used... - """ - - # create a mock response of a LemurPurgeResponse - mock_lemur_purge_response = factories.generate_dict_factory( - factories.LemurPurgeResponse - )() - - mock_request_id: str = str(uuid.uuid4()) - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR_BASE}/{mock_request_id}", - status_code=httpx.codes.OK, - method="DELETE", - json=mock_lemur_purge_response, - ) - - # mimic the usage of the SDK - result_future = aai.Lemur.purge_request_data_async(request_id=mock_request_id) - - result = result_future.result() - - # check the response - assert isinstance(result, aai.LemurPurgeResponse) - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_usage_data(httpx_mock: HTTPXMock): - """ - Tests whether usage data is correctly returned. - """ - - # create a mock response of a LemurTaskResponse - mock_lemur_task_response = factories.generate_dict_factory( - factories.LemurTaskResponse - )() - mock_lemur_task_response["usage"]["input_tokens"] = 100 - mock_lemur_task_response["usage"]["output_tokens"] = 200 - - # mock the specific endpoints - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR}/task", - status_code=httpx.codes.OK, - method="POST", - json=mock_lemur_task_response, - ) - - # mimic the usage of the SDK - transcript = aai.Transcript(str(uuid.uuid4())) - - lemur = aai.Lemur( - sources=[aai.LemurSource(transcript)], - ) - result = lemur.task(prompt="Create action items of the meeting") - - # check the response - assert isinstance(result, aai.LemurTaskResponse) - - assert ( - result.usage.input_tokens == mock_lemur_task_response["usage"]["input_tokens"] - ) - assert ( - result.usage.output_tokens == mock_lemur_task_response["usage"]["output_tokens"] - ) - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_get_response_data_string_response(httpx_mock: HTTPXMock): - """ - Tests whether a LeMUR string response data is correctly returned. - """ - request_id = "1234" - - mock_lemur_response = factories.generate_dict_factory( - factories.LemurStringResponse - )() - mock_lemur_response["request_id"] = request_id - - # mock the specific endpoint - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR_BASE}/{request_id}", - status_code=httpx.codes.OK, - method="GET", - json=mock_lemur_response, - ) - - # mimic the usage of the SDK - lemur = aai.Lemur() - result = lemur.get_response_data(request_id) - - # check the response - assert isinstance(result, aai.LemurStringResponse) - assert result.request_id == request_id - - # test the request field is populated correctly - assert result.request is not None - assert hasattr(result.request, "request_endpoint") - assert hasattr(result.request, "temperature") - assert hasattr(result.request, "final_model") - assert hasattr(result.request, "max_output_size") - assert hasattr(result.request, "created_at") - - # test usage field - assert result.usage is not None - assert hasattr(result.usage, "input_tokens") - assert hasattr(result.usage, "output_tokens") - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_get_response_data_question_answer(httpx_mock: HTTPXMock): - """ - Tests whether a LeMUR question-answer response data is correctly returned with questions field. - """ - request_id = "qa-1234" - - mock_lemur_response = factories.generate_dict_factory( - factories.LemurQuestionResponse - )() - mock_lemur_response["request"] = factories.generate_dict_factory( - factories.LemurQuestionRequestDetails - )() - mock_lemur_response["request_id"] = request_id - - # mock the specific endpoint - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR_BASE}/{request_id}", - status_code=httpx.codes.OK, - method="GET", - json=mock_lemur_response, - ) - - # mimic the usage of the SDK - lemur = aai.Lemur() - result = lemur.get_response_data(request_id) - - # check the response - assert isinstance(result, aai.LemurQuestionResponse) - assert result.request_id == request_id - - # test the request field is populated correctly - assert result.request is not None - assert result.request.request_endpoint == "/lemur/v3/question-answer" - assert hasattr(result.request, "temperature") - assert hasattr(result.request, "final_model") - assert hasattr(result.request, "max_output_size") - assert hasattr(result.request, "created_at") - - # test question-answer specific request fields - assert hasattr(result.request, "questions") - assert isinstance(result.request.questions, list) - assert len(result.request.questions) == 2 - - # test that questions have the right structure - one with answer_format, one with answer_options - question1, question2 = result.request.questions - assert hasattr(question1, "answer_format") - assert hasattr(question1, "context") - assert question1.question == "What is the main topic?" - assert question1.answer_format == "short sentence" - assert question1.context == "Meeting context" - - assert hasattr(question2, "answer_options") - assert question2.question == "What is the sentiment?" - assert question2.answer_options == ["positive", "negative", "neutral"] - - # test that qa-specific fields are None for other operation types - assert result.request.prompt is None # task-specific field - assert result.request.context is None # summary/action_items-specific field - assert result.request.answer_format is None # summary/action_items-specific field - - # test usage field - assert result.usage is not None - assert hasattr(result.usage, "input_tokens") - assert hasattr(result.usage, "output_tokens") - - # test response structure for question-answer - assert isinstance(result.response, list) - assert len(result.response) == 2 - for answer in result.response: - assert hasattr(answer, "question") - assert hasattr(answer, "answer") - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -@pytest.mark.parametrize("response_type", ("summary", "task")) -def test_lemur_get_response_data_additional_types(response_type, httpx_mock: HTTPXMock): - """ - Tests whether additional LeMUR response types are correctly returned with request details. - """ - request_id = "5678" - - # create a mock response - get_response_data returns LemurStringResponse but with different request details - mock_lemur_response = factories.generate_dict_factory( - factories.LemurStringResponse - )() - - # Override the request details based on response type - if response_type == "summary": - mock_lemur_response["request"] = factories.generate_dict_factory( - factories.LemurSummaryRequestDetails - )() - else: # task - mock_lemur_response["request"] = factories.generate_dict_factory( - factories.LemurTaskRequestDetails - )() - - mock_lemur_response["request_id"] = request_id - - # mock the specific endpoint - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR_BASE}/{request_id}", - status_code=httpx.codes.OK, - method="GET", - json=mock_lemur_response, - ) - - # mimic the usage of the SDK - lemur = aai.Lemur() - result = lemur.get_response_data(request_id) - - # check the response type - get_response_data returns LemurStringResponse for all string responses - assert isinstance(result, aai.LemurStringResponse) - - assert result.request_id == request_id - - # test the request field is populated correctly for all response types - assert result.request is not None - assert hasattr(result.request, "request_endpoint") - assert hasattr(result.request, "temperature") - assert hasattr(result.request, "final_model") - assert hasattr(result.request, "max_output_size") - assert hasattr(result.request, "created_at") - - # test type-specific request fields - if response_type == "summary": - assert result.request.request_endpoint == "/lemur/v3/summary" - assert result.request.context is not None - assert result.request.answer_format is not None - assert result.request.prompt is None # task-specific field - assert result.request.questions is None # qa-specific field - else: # task - assert result.request.request_endpoint == "/lemur/v3/task" - assert result.request.prompt is not None - assert result.request.context is None # summary-specific field - assert result.request.answer_format is None # summary-specific field - assert result.request.questions is None # qa-specific field - - # test usage field - assert result.usage is not None - assert hasattr(result.usage, "input_tokens") - assert hasattr(result.usage, "output_tokens") - - # test string response field - assert isinstance(result.response, str) - - # check whether we mocked everything - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_get_response_data_with_optional_request_fields(httpx_mock: HTTPXMock): - """ - Tests that optional fields in LemurRequestDetails are handled correctly. - """ - request_id = "test-optional-fields" - - mock_lemur_response = factories.generate_dict_factory( - factories.LemurStringResponse - )() - mock_lemur_response["request_id"] = request_id - - # Add optional fields to request details - mock_lemur_response["request"]["transcript_ids"] = ["transcript_1", "transcript_2"] - mock_lemur_response["request"]["input_text"] = "Test input text" - mock_lemur_response["request"]["questions"] = [ - {"question": "What is this about?", "answer_format": "short"} - ] - mock_lemur_response["request"]["prompt"] = "Test prompt" - mock_lemur_response["request"]["context"] = {"key": "value"} - mock_lemur_response["request"]["answer_format"] = "bullet points" - - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR_BASE}/{request_id}", - status_code=httpx.codes.OK, - method="GET", - json=mock_lemur_response, - ) - - lemur = aai.Lemur() - result = lemur.get_response_data(request_id) - - assert isinstance(result, aai.LemurStringResponse) - assert result.request_id == request_id - - # test that optional fields are present - assert result.request.transcript_ids == ["transcript_1", "transcript_2"] - assert result.request.input_text == "Test input text" - assert result.request.questions is not None - assert result.request.prompt == "Test prompt" - assert result.request.context == {"key": "value"} - assert result.request.answer_format == "bullet points" - - assert len(httpx_mock.get_requests()) == 1 - - -def test_lemur_get_response_data_fails(httpx_mock: HTTPXMock): - """ - Tests that get_response_data properly handles API errors. - """ - request_id = "error-request-id" - - httpx_mock.add_response( - url=f"{aai.settings.base_url}{ENDPOINT_LEMUR_BASE}/{request_id}", - status_code=httpx.codes.NOT_FOUND, - method="GET", - json={"error": "Request not found"}, - ) - - lemur = aai.Lemur() - - with pytest.raises(aai.LemurError): - lemur.get_response_data(request_id) - - assert len(httpx_mock.get_requests()) == 1 diff --git a/tests/unit/test_sync.py b/tests/unit/test_sync.py index e41e1cb..fa1a655 100644 --- a/tests/unit/test_sync.py +++ b/tests/unit/test_sync.py @@ -89,7 +89,7 @@ def test_transcribe_sends_prompt_and_keyterms_prompt(httpx_mock: HTTPXMock): # When transcribing with a prompt and keyterms_prompt config = aai.SyncTranscriptionConfig( prompt="Transcribe verbatim.", - keyterms_prompt=["AssemblyAI", " Lemur ", ""], + keyterms_prompt=["AssemblyAI", " Universal ", ""], ) aai.SyncTranscriber().transcribe(b"RIFFfake-wav-bytes", config=config) @@ -98,7 +98,7 @@ def test_transcribe_sends_prompt_and_keyterms_prompt(httpx_mock: HTTPXMock): assert b'name="config"' in body assert b"Transcribe verbatim." in body assert b'"AssemblyAI"' in body - assert b'"Lemur"' in body # whitespace stripped, empty term dropped + assert b'"Universal"' in body # whitespace stripped, empty term dropped # And the routing model is never placed in the body assert b'"model"' not in body diff --git a/tests/unit/test_transcriber_backwards_compat.py b/tests/unit/test_transcriber_backwards_compat.py index 14ac415..b6407e9 100644 --- a/tests/unit/test_transcriber_backwards_compat.py +++ b/tests/unit/test_transcriber_backwards_compat.py @@ -92,7 +92,7 @@ def test_root_api_module_reexports_prerecorded_endpoints(): f"assemblyai.api.{name} does not match prerecorded.v2.api.{name}" ) - for name in ("ENDPOINT_UPLOAD", "upload_file", "lemur_task", "_get_error_message"): + for name in ("ENDPOINT_UPLOAD", "upload_file", "_get_error_message"): assert hasattr(root_api, name), f"assemblyai.api.{name} is gone"