13040707031

Committed 29 Jan 2025 09:23PM UTC coverage: 96.412% (+15.9%) from 80.557%

Build # 13040707031

Build Type

Pull #84

github

Specific Base 24322d

Committed by

kevdevg

Commit Message

feat: multimodal output for openain/litellm

Pull Request Pull Request #84: feat: multimodal output audio for OpenAi and Litellm

Run Details

33 of 34 new or added lines in 3 files covered. (97.06%)

54 existing lines in 10 files now uncovered.

2472 of 2564 relevant lines covered (96.41%)

3.85 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.25

/scope3ai/tracers/huggingface/speech_to_text.py

import time
from dataclasses import asdict, dataclass
from typing import Any, Callable, Optional, Tuple, Union

from aiohttp import ClientResponse
from huggingface_hub import (
    AsyncInferenceClient,
    InferenceClient,  # type: ignore[import-untyped]
)
from huggingface_hub import (
    AutomaticSpeechRecognitionOutput as _AutomaticSpeechRecognitionOutput,
)
from requests import Response

from scope3ai.api.types import ImpactRow, Scope3AIContext
from scope3ai.api.typesgen import Task
from scope3ai.constants import PROVIDERS
from scope3ai.lib import Scope3AI
from scope3ai.response_interceptor.aiohttp_interceptor import aiohttp_response_capture
from scope3ai.response_interceptor.requests_interceptor import requests_response_capture

PROVIDER = PROVIDERS.HUGGINGFACE_HUB.value
HUGGING_FACE_SPEECH_TO_TEXT_TASK = "automatic-speech-recognition"


@dataclass
class AutomaticSpeechRecognitionOutput(_AutomaticSpeechRecognitionOutput):
    scope3ai: Optional[Scope3AIContext] = None


def _hugging_face_automatic_recognition_get_impact_row(
    timer_start: Any,
    model: Any,
    response: Any,
    http_response: Optional[Union[ClientResponse, Response]],
    args: Any,
    kwargs: Any,
) -> Tuple[AutomaticSpeechRecognitionOutput, ImpactRow]:
    compute_audio_length = None
    compute_time = 0
    if http_response:
        compute_audio_length = http_response.headers.get("x-compute-audio-length")
        compute_time = http_response.headers.get("x-compute-time")
    if not compute_time:
        compute_time = time.perf_counter() - timer_start
    if not compute_audio_length:
        compute_audio_length = 0
    scope3_row = ImpactRow(
        model_id=model,
        task=Task.text_to_speech,
        input_audio_seconds=float(compute_audio_length),
        request_duration_ms=float(compute_time) * 1000,
        managed_service_id=PROVIDER,
    )
    result = AutomaticSpeechRecognitionOutput(**asdict(response))
    return result, scope3_row


def huggingface_automatic_recognition_output_wrapper(
    wrapped: Callable, instance: InferenceClient, args: Any, kwargs: Any
) -> AutomaticSpeechRecognitionOutput:
    timer_start = time.perf_counter()
    http_response: Response | None = None
    with requests_response_capture() as responses:
        response = wrapped(*args, **kwargs)
        http_responses = responses.get()
        if http_responses:
            http_response = http_responses[-1]
    model = kwargs.get("model") or instance.get_recommended_model(
        HUGGING_FACE_SPEECH_TO_TEXT_TASK
    )
    result, impact_row = _hugging_face_automatic_recognition_get_impact_row(
        timer_start, model, response, http_response, args, kwargs
    )
    scope3_ctx = Scope3AI.get_instance().submit_impact(impact_row)
    result.scope3ai = scope3_ctx
    return result


async def huggingface_automatic_recognition_output_wrapper_async(
    wrapped: Callable, instance: AsyncInferenceClient, args: Any, kwargs: Any
) -> AutomaticSpeechRecognitionOutput:
    timer_start = time.perf_counter()
    http_response: ClientResponse | None = None
    with aiohttp_response_capture() as responses:
        response = await wrapped(*args, **kwargs)
        http_responses = responses.get()
        if http_responses:
            http_response = http_responses[-1]
    model = kwargs.get("model") or instance.get_recommended_model(
        HUGGING_FACE_SPEECH_TO_TEXT_TASK
    )
    result, impact_row = _hugging_face_automatic_recognition_get_impact_row(
        timer_start, model, response, http_response, args, kwargs
    )
    scope3_ctx = await Scope3AI.get_instance().asubmit_impact(impact_row)
    result.scope3ai = scope3_ctx
    return result

1	import time	4✔
2	from dataclasses import asdict, dataclass	4✔
3	from typing import Any, Callable, Optional, Tuple, Union	4✔
4
5	from aiohttp import ClientResponse	4✔
6	from huggingface_hub import (	4✔
7	AsyncInferenceClient,
8	InferenceClient, # type: ignore[import-untyped]
9	)
10	from huggingface_hub import (	4✔
11	AutomaticSpeechRecognitionOutput as _AutomaticSpeechRecognitionOutput,
12	)
13	from requests import Response	4✔
14
15	from scope3ai.api.types import ImpactRow, Scope3AIContext	4✔
16	from scope3ai.api.typesgen import Task	4✔
17	from scope3ai.constants import PROVIDERS	4✔
18	from scope3ai.lib import Scope3AI	4✔
19	from scope3ai.response_interceptor.aiohttp_interceptor import aiohttp_response_capture	4✔
20	from scope3ai.response_interceptor.requests_interceptor import requests_response_capture	4✔
21
22	PROVIDER = PROVIDERS.HUGGINGFACE_HUB.value	4✔
23	HUGGING_FACE_SPEECH_TO_TEXT_TASK = "automatic-speech-recognition"	4✔
24
25
26	@dataclass	4✔
27	class AutomaticSpeechRecognitionOutput(_AutomaticSpeechRecognitionOutput):	4✔
28	scope3ai: Optional[Scope3AIContext] = None	4✔
29
30
31	def _hugging_face_automatic_recognition_get_impact_row(	4✔
32	timer_start: Any,
33	model: Any,
34	response: Any,
35	http_response: Optional[Union[ClientResponse, Response]],
36	args: Any,
37	kwargs: Any,
38	) -> Tuple[AutomaticSpeechRecognitionOutput, ImpactRow]:
39	compute_audio_length = None	4✔
40	compute_time = 0	4✔
41	if http_response:	4✔
42	compute_audio_length = http_response.headers.get("x-compute-audio-length")	4✔
43	compute_time = http_response.headers.get("x-compute-time")	4✔
44	if not compute_time:	4✔
UNCOV 45	compute_time = time.perf_counter() - timer_start	×
46	if not compute_audio_length:	4✔
47	compute_audio_length = 0	4✔
48	scope3_row = ImpactRow(	4✔
49	model_id=model,
50	task=Task.text_to_speech,
51	input_audio_seconds=float(compute_audio_length),
52	request_duration_ms=float(compute_time) * 1000,
53	managed_service_id=PROVIDER,
54	)
55	result = AutomaticSpeechRecognitionOutput(**asdict(response))	4✔
56	return result, scope3_row	4✔
57
58
59	def huggingface_automatic_recognition_output_wrapper(	4✔
60	wrapped: Callable, instance: InferenceClient, args: Any, kwargs: Any
61	) -> AutomaticSpeechRecognitionOutput:
62	timer_start = time.perf_counter()	4✔
63	http_response: Response \| None = None	4✔
64	with requests_response_capture() as responses:	4✔
65	response = wrapped(args, *kwargs)	4✔
66	http_responses = responses.get()	4✔
67	if http_responses:	4✔
68	http_response = http_responses[-1]	4✔
69	model = kwargs.get("model") or instance.get_recommended_model(	4✔
70	HUGGING_FACE_SPEECH_TO_TEXT_TASK
71	)
72	result, impact_row = _hugging_face_automatic_recognition_get_impact_row(	4✔
73	timer_start, model, response, http_response, args, kwargs
74	)
75	scope3_ctx = Scope3AI.get_instance().submit_impact(impact_row)	4✔
76	result.scope3ai = scope3_ctx	4✔
77	return result	4✔
78
79
80	async def huggingface_automatic_recognition_output_wrapper_async(	4✔
81	wrapped: Callable, instance: AsyncInferenceClient, args: Any, kwargs: Any
82	) -> AutomaticSpeechRecognitionOutput:
83	timer_start = time.perf_counter()	4✔
84	http_response: ClientResponse \| None = None	4✔
85	with aiohttp_response_capture() as responses:	4✔
86	response = await wrapped(args, *kwargs)	4✔
87	http_responses = responses.get()	4✔
88	if http_responses:	4✔
89	http_response = http_responses[-1]	4✔
90	model = kwargs.get("model") or instance.get_recommended_model(	4✔
91	HUGGING_FACE_SPEECH_TO_TEXT_TASK
92	)
93	result, impact_row = _hugging_face_automatic_recognition_get_impact_row(	4✔
94	timer_start, model, response, http_response, args, kwargs
95	)
96	scope3_ctx = await Scope3AI.get_instance().asubmit_impact(impact_row)	4✔
97	result.scope3ai = scope3_ctx	4✔
98	return result	4✔

scope3data / scope3ai-py / 13040707031

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous