fix: fix async submit impact

scope3ai/tracers/anthropic/chat.py

@@ -108,7 +108,7 @@ async def __stream_text__(self) -> AsyncIterator[str]:  # type: ignore[misc]
                 request_duration_ms=requests_latency * 1000,
                 managed_service_id=PROVIDER,
             )
-            self.scope3ai = Scope3AI.get_instance().submit_impact(scope3_row)
+            self.scope3ai = await Scope3AI.get_instance().asubmit_impact(scope3_row)
 
     def __init__(self, parent) -> None:  # noqa: ANN001
         super().__init__(
@@ -212,7 +212,7 @@ async def __stream__(self) -> AsyncIterator[_T]:
             request_duration_ms=request_latency * 1000,
             managed_service_id=PROVIDER,
         )
-        self.scope3ai = Scope3AI.get_instance().submit_impact(scope3_row)
+        self.scope3ai = await Scope3AI.get_instance().asubmit_impact(scope3_row)
 
     def __init__(self, parent) -> None:  # noqa: ANN001
         super().__init__(
@@ -265,7 +265,7 @@ async def _anthropic_async_chat_wrapper(
         request_duration_ms=request_latency * 1000,
         managed_service_id=PROVIDER,
     )
-    scope3ai_ctx = Scope3AI.get_instance().submit_impact(scope3_row)
+    scope3ai_ctx = await Scope3AI.get_instance().asubmit_impact(scope3_row)
     if scope3ai_ctx is not None:
         return Message(**response.model_dump(), scope3ai=scope3ai_ctx)
     else:

scope3ai/tracers/cohere/chat.py

@@ -70,7 +70,7 @@ async def cohere_async_chat_wrapper(
         request_duration_ms=request_latency * 1000,
         managed_service_id=PROVIDER,
     )
-    scope3ai_ctx = Scope3AI.get_instance().submit_impact(scope3_row)
+    scope3ai_ctx = await Scope3AI.get_instance().asubmit_impact(scope3_row)
     return NonStreamedChatResponse(**response.dict(), scope3ai=scope3ai_ctx)
 
 
@@ -122,7 +122,7 @@ async def cohere_async_stream_chat_wrapper(
                 request_duration_ms=request_latency * 1000,
                 managed_service_id=PROVIDER,
             )
-            scope3ai_ctx = Scope3AI.get_instance().submit_impact(scope3_row)
+            scope3ai_ctx = await Scope3AI.get_instance().asubmit_impact(scope3_row)
             yield StreamEndStreamedChatResponse(**event.dict(), scope3ai=scope3ai_ctx)
         else:
             yield event

scope3ai/tracers/cohere/chat_v2.py

@@ -73,7 +73,7 @@ async def cohere_async_chat_v2_wrapper(
         request_duration_ms=request_latency * 1000,
         managed_service_id=PROVIDER,
     )
-    scope3ai_ctx = Scope3AI.get_instance().submit_impact(scope3_row)
+    scope3ai_ctx = await Scope3AI.get_instance().asubmit_impact(scope3_row)
     return ChatResponse(**response.dict(), scope3ai=scope3ai_ctx)
 
 
@@ -125,5 +125,5 @@ async def cohere_async_stream_chat_v2_wrapper(
                 request_duration_ms=request_latency * 1000,
                 managed_service_id=PROVIDER,
             )
-            scope3ai_ctx = Scope3AI.get_instance().submit_impact(scope3_row)
+            scope3ai_ctx = await Scope3AI.get_instance().asubmit_impact(scope3_row)
             yield Scope3AIStreamedChatResponseV2(type="scope3ai", scope3ai=scope3ai_ctx)

scope3ai/tracers/litellm/chat.py

@@ -114,7 +114,7 @@ async def litellm_async_chat_wrapper_base(
         request_duration_ms=float(request_latency) * 1000,
         managed_service_id=PROVIDER,
     )
-    scope3ai_ctx = Scope3AI.get_instance().submit_impact(scope3_row)
+    scope3ai_ctx = await Scope3AI.get_instance().asubmit_impact(scope3_row)
     if scope3ai_ctx is not None:
         return ChatCompletion(**response.model_dump(), scope3ai=scope3ai_ctx)
     else:
@@ -143,7 +143,7 @@ async def litellm_async_chat_wrapper_stream(  # type: ignore[misc]
                 request_duration_ms=float(request_latency) * 1000,
                 managed_service_id=PROVIDER,
             )
-            scope3ai_ctx = Scope3AI.get_instance().submit_impact(scope3_row)
+            scope3ai_ctx = await Scope3AI.get_instance().asubmit_impact(scope3_row)
             if scope3ai_ctx is not None:
                 yield ChatCompletionChunk(**chunk.model_dump(), scope3ai=scope3ai_ctx)
             else:

scope3ai/tracers/mistralai/chat.py

@@ -65,7 +65,7 @@ def mistralai_v1_chat_wrapper_stream(
                 input_tokens=token_count,
                 output_tokens=chunk.data.usage.completion_tokens
                 if chunk.data.usage
-                else None,
+                else 0,
                 request_duration_ms=request_latency * 1000,
                 managed_service_id=PROVIDER,
             )
@@ -92,7 +92,7 @@ async def mistralai_v1_async_chat_wrapper(
         request_duration_ms=request_latency * 1000,
         managed_service_id=PROVIDER,
     )
-    scope3ai_ctx = Scope3AI.get_instance().submit_impact(scope3_row)
+    scope3ai_ctx = await Scope3AI.get_instance().asubmit_impact(scope3_row)
     chat = ChatCompletionResponse(**response.model_dump())
     chat.scope3ai = scope3ai_ctx
     return chat
@@ -110,13 +110,11 @@ async def _generator(
         scope3_row = ImpactRow(
             model_id=model_name,
             input_tokens=token_count,
-            output_tokens=chunk.data.usage.completion_tokens
-            if chunk.data.usage
-            else None,
+            output_tokens=chunk.data.usage.completion_tokens if chunk.data.usage else 0,
             request_duration_ms=request_latency * 1000,
             managed_service_id=PROVIDER,
         )
-        scope3ai_ctx = Scope3AI.get_instance().submit_impact(scope3_row)
+        scope3ai_ctx = await Scope3AI.get_instance().asubmit_impact(scope3_row)
         chunk.data = CompletionChunk(**chunk.data.model_dump(), scope3ai=scope3ai_ctx)
         yield chunk
 

scope3ai/tracers/openai/chat.py

@@ -60,7 +60,8 @@ def _openai_aggregate_multimodal_audio(content: dict, row: ImpactRow) -> None:
 
     # decode the base64 data
     audio_data = base64.b64decode(b64data)
-    duration = _get_audio_duration(format, audio_data)
+    # TODO: accept audio duration as float in AiApi
+    duration = int(_get_audio_duration(format, audio_data))
 
     if row.input_audio_seconds is None:
         row.input_audio_seconds = duration
@@ -107,7 +108,6 @@ def _openai_chat_wrapper(
     messages = kwargs.get("messages", [])
     for message in messages:
         _openai_aggregate_multimodal(message, scope3_row)
-
     scope3ai_ctx = Scope3AI.get_instance().submit_impact(scope3_row)
     return ChatCompletion(**response.model_dump(), scope3ai=scope3ai_ctx)
 
@@ -208,7 +208,7 @@ async def openai_async_chat_wrapper_stream(
                 managed_service_id=PROVIDER,
             )
 
-            scope3_ctx = Scope3AI.get_instance().submit_impact(scope3_row)
+            scope3_ctx = await Scope3AI.get_instance().asubmit_impact(scope3_row)
             yield ChatCompletionChunk(**chunk.model_dump(), scope3ai=scope3_ctx)
         else:
             yield chunk

scope3ai/tracers/openai/speech_to_text.py

@@ -32,9 +32,9 @@ class TranscriptionVerbose(_TranscriptionVerbose):
     scope3ai: Optional[Scope3AIContext] = None
 
 
-def _openai_speech_to_text_wrapper(
+def _openai_speech_to_text_get_impact_row(
     response: Any, request_latency: float, kwargs: dict
-) -> Union[Transcription, TranscriptionVerbose, str]:
+) -> (Union[Transcription, TranscriptionVerbose, str], ImpactRow):
     model = kwargs["model"]
     encoder = tiktoken.get_encoding("cl100k_base")
 
@@ -58,7 +58,6 @@ def _openai_speech_to_text_wrapper(
         task=Task.speech_to_text,
         **options,
     )
-    scope3_ctx = Scope3AI.get_instance().submit_impact(scope3_row)
 
     if isinstance(response, _Transcription):
         result = Transcription.model_construct(**response.model_dump())
@@ -69,8 +68,7 @@ def _openai_speech_to_text_wrapper(
     else:
         logger.error(f"Unexpected response type: {type(response)}")
         return response
-    result.scope3ai = scope3_ctx
-    return result
+    return result, scope3_row
 
 
 def openai_speech_to_text_wrapper(
@@ -79,7 +77,12 @@ def openai_speech_to_text_wrapper(
     timer_start = time.perf_counter()
     response = wrapped(*args, **kwargs)
     request_latency = (time.perf_counter() - timer_start) * 1000
-    return _openai_speech_to_text_wrapper(response, request_latency, kwargs)
+    result, impact_row = _openai_speech_to_text_get_impact_row(
+        response, request_latency, kwargs
+    )
+    scope3_ctx = Scope3AI.get_instance().submit_impact(impact_row)
+    result.scope3ai = scope3_ctx
+    return result
 
 
 async def openai_async_speech_to_text_wrapper(
@@ -88,4 +91,9 @@ async def openai_async_speech_to_text_wrapper(
     timer_start = time.perf_counter()
     response = await wrapped(*args, **kwargs)
     request_latency = (time.perf_counter() - timer_start) * 1000
-    return _openai_speech_to_text_wrapper(response, request_latency, kwargs)
+    result, impact_row = _openai_speech_to_text_get_impact_row(
+        response, request_latency, kwargs
+    )
+    scope3_ctx = await Scope3AI.get_instance().asubmit_impact(impact_row)
+    result.scope3ai = scope3_ctx
+    return result

scope3ai/tracers/openai/text_to_image.py

@@ -18,9 +18,9 @@ class ImageResponse(_ImageResponse):
     scope3ai: Optional[Scope3AIContext] = None
 
 
-def _openai_image_wrapper(
+def _openai_image_get_impact_row(
     response: _ImageResponse, request_latency: float, **kwargs: Any
-) -> ImageResponse:
+) -> (ImageResponse, ImpactRow):
     model = kwargs.get("model", DEFAULT_MODEL)
     size = RootImage(root=kwargs.get("size", DEFAULT_SIZE))
     n = kwargs.get("n", DEFAULT_N)
@@ -33,10 +33,8 @@ def _openai_image_wrapper(
         managed_service_id=PROVIDER,
     )
 
-    scope3ai_ctx = Scope3AI.get_instance().submit_impact(scope3_row)
     result = ImageResponse.model_construct(**response.model_dump())
-    result.scope3ai = scope3ai_ctx
-    return result
+    return result, scope3_row
 
 
 def openai_image_wrapper(
@@ -45,7 +43,12 @@ def openai_image_wrapper(
     timer_start = time.perf_counter()
     response = wrapped(*args, **kwargs)
     request_latency = time.perf_counter() - timer_start
-    return _openai_image_wrapper(response, request_latency, **kwargs)
+    result, impact_row = _openai_image_get_impact_row(
+        response, request_latency, **kwargs
+    )
+    scope3_ctx = Scope3AI.get_instance().submit_impact(impact_row)
+    result.scope3ai = scope3_ctx
+    return result
 
 
 async def openai_async_image_wrapper(
@@ -54,4 +57,9 @@ async def openai_async_image_wrapper(
     timer_start = time.perf_counter()
     response = await wrapped(*args, **kwargs)
     request_latency = time.perf_counter() - timer_start
-    return _openai_image_wrapper(response, request_latency, **kwargs)
+    result, impact_row = _openai_image_get_impact_row(
+        response, request_latency, **kwargs
+    )
+    scope3_ctx = await Scope3AI.get_instance().asubmit_impact(impact_row)
+    result.scope3ai = scope3_ctx
+    return result

scope3ai/tracers/openai/text_to_speech.py

@@ -20,11 +20,11 @@ class HttpxBinaryResponseContent(_legacy_response.HttpxBinaryResponseContent):
     scope3ai: Optional[Scope3AIContext] = None
 
 
-def _openai_text_to_speech_submit(
+def _openai_text_to_speech_get_impact_row(
     response: _legacy_response.HttpxBinaryResponseContent,
     request_latency: float,
     kwargs: Any,
-) -> HttpxBinaryResponseContent:
+) -> (HttpxBinaryResponseContent, ImpactRow):
     # try getting duration
     response_format = kwargs["response_format"]
     duration = _get_audio_duration(response_format, response.content)
@@ -49,13 +49,10 @@ def _openai_text_to_speech_submit(
         task=Task.text_to_speech,
     )
 
-    scope3_ctx = Scope3AI.get_instance().submit_impact(scope3_row)
-
     wrapped_response = HttpxBinaryResponseContent(
         response=response.response,
     )
-    wrapped_response.scope3ai = scope3_ctx
-    return wrapped_response
+    return wrapped_response, scope3_row
 
 
 def openai_text_to_speech_wrapper(
@@ -64,13 +61,23 @@ def openai_text_to_speech_wrapper(
     timer_start = time.perf_counter()
     response = wrapped(*args, **kwargs)
     request_latency = (time.perf_counter() - timer_start) * 1000
-    return _openai_text_to_speech_submit(response, request_latency, kwargs)
+    result, impact_row = _openai_text_to_speech_get_impact_row(
+        response, request_latency, kwargs
+    )
+    scope3_ctx = Scope3AI.get_instance().submit_impact(impact_row)
+    result.scope3ai = scope3_ctx
+    return result
 
 
 async def openai_async_text_to_speech_wrapper(
     wrapped: Callable, instance: AsyncSpeech, args: Any, kwargs: Any
 ) -> HttpxBinaryResponseContent:
     timer_start = time.perf_counter()
     response = await wrapped(*args, **kwargs)
-    request_latency = time.perf_counter() - timer_start
-    return _openai_text_to_speech_submit(response, request_latency, kwargs)
+    request_latency = (time.perf_counter() - timer_start) * 1000
+    result, impact_row = _openai_text_to_speech_get_impact_row(
+        response, request_latency, kwargs
+    )
+    scope3_ctx = await Scope3AI.get_instance().asubmit_impact(impact_row)
+    result.scope3ai = scope3_ctx
+    return result

scope3ai/tracers/openai/translation.py

@@ -1,6 +1,6 @@
 import logging
 import time
-from typing import Any, Callable, Optional, Union
+from typing import Any, Callable, Optional, Union, Tuple
 
 import tiktoken
 from openai.resources.audio.translations import AsyncTranslations, Translations
@@ -32,9 +32,9 @@ class TranslationVerbose(_TranslationVerbose):
     scope3ai: Optional[Scope3AIContext] = None
 
 
-def _openai_translation_wrapper(
+def _openai_translation_get_impact_row(
     response: Any, request_latency: float, kwargs: dict
-) -> Union[Translation, TranslationVerbose, AnnotatedStr]:
+) -> Tuple[Union[Translation, TranslationVerbose, AnnotatedStr], ImpactRow]:
     model = kwargs["model"]
     encoder = tiktoken.get_encoding("cl100k_base")
 
@@ -58,7 +58,6 @@ def _openai_translation_wrapper(
         task=Task.translation,
         **options,
     )
-    scope3_ctx = Scope3AI.get_instance().submit_impact(scope3_row)
 
     if isinstance(response, _Translation):
         result = Translation.model_construct(**response.model_dump())
@@ -68,9 +67,8 @@ def _openai_translation_wrapper(
         result = AnnotatedStr(str)
     else:
         logger.error(f"Unexpected response type: {type(response)}")
-        return response
-    result.scope3ai = scope3_ctx
-    return result
+        return response, scope3_row
+    return result, scope3_row
 
 
 def openai_translation_wrapper(
@@ -79,7 +77,12 @@ def openai_translation_wrapper(
     timer_start = time.perf_counter()
     response = wrapped(*args, **kwargs)
     request_latency = (time.perf_counter() - timer_start) * 1000
-    return _openai_translation_wrapper(response, request_latency, kwargs)
+    result, impact_row = _openai_translation_get_impact_row(
+        response, request_latency, kwargs
+    )
+    scope3_ctx = Scope3AI.get_instance().submit_impact(impact_row)
+    result.scope3ai = scope3_ctx
+    return result
 
 
 async def openai_async_translation_wrapper(
@@ -88,4 +91,9 @@ async def openai_async_translation_wrapper(
     timer_start = time.perf_counter()
     response = await wrapped(*args, **kwargs)
     request_latency = (time.perf_counter() - timer_start) * 1000
-    return _openai_translation_wrapper(response, request_latency, kwargs)
+    result, impact_row = _openai_translation_get_impact_row(
+        response, request_latency, kwargs
+    )
+    scope3_ctx = await Scope3AI.get_instance().asubmit_impact(impact_row)
+    result.scope3ai = scope3_ctx
+    return result