Merge pull request #1 from liberate-org/exp/vad

added Silerio VAD and updated OpenAI to latest SDK

Author: liberate-eng

.gitignore

@@ -11,3 +11,4 @@ benchmark_results/
 private.key
 dump.rdb
 .idea
+.venv

poetry.lock

@@ -11,7 +11,7 @@ homepage = "https://github.com/vocodedev/vocode-python"
 python = ">=3.8.1,<3.12"
 pydub = "^0.25.1"
 nltk = "^3.8.1"
-openai = "^0.27.8"
+openai = "1.12.0"
 sounddevice = "^0.4.6"
 azure-cognitiveservices-speech = "^1.27.0"
 websockets = "^11.0.2"
@@ -42,6 +42,9 @@ langchain = "^0.0.198"
 google-cloud-aiplatform = {version = "^1.26.0", optional = true}
 miniaudio = "^1.59"
 boto3 = "^1.28.28"
+pandas = "2.0.3"
+torch = "2.1.1"
+torchaudio = "2.1.1"
 
 
 [tool.poetry.group.lint.dependencies]

pyproject.toml

@@ -2,7 +2,12 @@
 
 from typing import Any, Dict, List, Optional, Tuple, Union
 
-import openai
+from openai import (
+    OpenAI,
+    AsyncOpenAI,
+    AsyncAzureOpenAI,
+    AzureOpenAI,
+)
 from typing import AsyncGenerator, Optional, Tuple
 
 import logging
@@ -37,16 +42,26 @@ def __init__(
             agent_config=agent_config, action_factory=action_factory, logger=logger
         )
         if agent_config.azure_params:
-            openai.api_type = agent_config.azure_params.api_type
-            openai.api_base = getenv("AZURE_OPENAI_API_BASE")
-            openai.api_version = agent_config.azure_params.api_version
-            openai.api_key = getenv("AZURE_OPENAI_API_KEY")
+            self.openaiAsyncClient = AsyncAzureOpenAI(
+                api_version = agent_config.azure_params.api_version,
+                base_url = getenv("AZURE_OPENAI_API_BASE"),
+                api_key = getenv("AZURE_OPENAI_API_KEY")
+            )
+            self.openaiSyncClient = AzureOpenAI(
+                api_version = agent_config.azure_params.api_version,
+                base_url = getenv("AZURE_OPENAI_API_BASE"),
+                api_key = getenv("AZURE_OPENAI_API_KEY")
+            )
         else:
-            openai.api_type = "open_ai"
-            openai.api_base = "https://api.openai.com/v1"
-            openai.api_version = None
-            openai.api_key = openai_api_key or getenv("OPENAI_API_KEY")
-        if not openai.api_key:
+            self.openaiAsyncClient = AsyncOpenAI(
+                base_url = "https://api.openai.com/v1",
+                api_key = openai_api_key or getenv("OPENAI_API_KEY")
+            )
+            self.openaiSyncClient = OpenAI(
+                base_url = "https://api.openai.com/v1",
+                api_key = openai_api_key or getenv("OPENAI_API_KEY")
+            )
+        if not self.openaiAsyncClient.api_key or not self.openaiSyncClient.api_key:
             raise ValueError("OPENAI_API_KEY must be set in environment or passed in")
         self.first_response = (
             self.create_first_response(agent_config.expected_first_prompt)
@@ -104,7 +119,7 @@ def create_first_response(self, first_prompt):
         ]
 
         parameters = self.get_chat_parameters(messages)
-        return openai.ChatCompletion.create(**parameters)
+        return self.openaiSyncClient.chat.completions.create(**parameters)
 
     def attach_transcript(self, transcript: Transcript):
         self.transcript = transcript
@@ -126,7 +141,8 @@ async def respond(
             text = self.first_response
         else:
             chat_parameters = self.get_chat_parameters()
-            chat_completion = await openai.ChatCompletion.acreate(**chat_parameters)
+            # chat_completion = await openai.ChatCompletion.acreate(**chat_parameters)
+            chat_completion = await self.openaiAsyncClient.chat.completions.create(**chat_parameters)
             text = chat_completion.choices[0].message.content
         self.logger.debug(f"LLM response: {text}")
         return text, False
@@ -172,7 +188,8 @@ async def generate_response(
         else:
             chat_parameters = self.get_chat_parameters()
         chat_parameters["stream"] = True
-        stream = await openai.ChatCompletion.acreate(**chat_parameters)
+        # stream = await openai.ChatCompletion.acreate(**chat_parameters)
+        stream = await self.openaiAsyncClient.chat.completions.create(**chat_parameters)
         async for message in collate_response_async(
             openai_get_tokens(stream), get_functions=True
         ):

vocode/streaming/agent/chat_gpt_agent.py

@@ -1,5 +1,6 @@
 from copy import deepcopy
 import re
+import time
 from typing import (
     Dict,
     Any,
@@ -12,8 +13,8 @@
     TypeVar,
     Union,
 )
+import logging
 
-from openai.openai_object import OpenAIObject
 from vocode.streaming.models.actions import FunctionCall, FunctionFragment
 from vocode.streaming.models.events import Sender
 from vocode.streaming.models.transcript import (
@@ -31,6 +32,8 @@ async def collate_response_async(
     gen: AsyncIterable[Union[str, FunctionFragment]],
     sentence_endings: List[str] = SENTENCE_ENDINGS,
     get_functions: Literal[True, False] = False,
+    logger: Optional[logging.Logger] = None,
+    start_token_processing: Optional[float] = time.time()
 ) -> AsyncGenerator[Union[str, FunctionCall], None]:
     sentence_endings_pattern = "|".join(map(re.escape, sentence_endings))
     list_item_ending_pattern = r"\n"
@@ -43,6 +46,10 @@ async def collate_response_async(
             continue
         if isinstance(token, str):
             if prev_ends_with_money and token.startswith(" "):
+                if logger:
+                    logger.debug("Took %s to generate [%s]", 
+                                 time.time() - start_token_processing, 
+                                 buffer.strip())
                 yield buffer.strip()
                 buffer = ""
 
@@ -58,6 +65,10 @@ async def collate_response_async(
                 if not ends_with_money:
                     to_return = buffer.strip()
                     if to_return:
+                        if logger:
+                            logger.debug("Took %s to generate [%s]",
+                              time.time() - start_token_processing,
+                              to_return)
                         yield to_return
                     buffer = ""
             prev_ends_with_money = ends_with_money
@@ -66,35 +77,47 @@ async def collate_response_async(
             function_args_buffer += token.arguments
     to_return = buffer.strip()
     if to_return:
+        if logger:
+            logger.debug("Took %s to generate [%s]",
+                time.time() - start_token_processing,
+                to_return)
         yield to_return
     if function_name_buffer and get_functions:
         yield FunctionCall(name=function_name_buffer, arguments=function_args_buffer)
 
 
 async def openai_get_tokens(gen) -> AsyncGenerator[Union[str, FunctionFragment], None]:
     async for event in gen:
-        choices = event.get("choices", [])
+        choices = event.choices or []
         if len(choices) == 0:
-            continue
+            break
         choice = choices[0]
         if choice.finish_reason:
             break
-        delta = choice.get("delta", {})
-        if "text" in delta and delta["text"] is not None:
-            token = delta["text"]
+        delta = choice.delta or {}
+        if hasattr(delta, "text") and delta.text:
+            token = delta.text
             yield token
-        if "content" in delta and delta["content"] is not None:
-            token = delta["content"]
+        if hasattr(delta, "content") and delta.content:
+            token = delta.content
             yield token
-        elif "function_call" in delta and delta["function_call"] is not None:
-            yield FunctionFragment(
-                name=delta["function_call"]["name"]
-                if "name" in delta["function_call"]
-                else "",
-                arguments=delta["function_call"]["arguments"]
-                if "arguments" in delta["function_call"]
-                else "",
-            )
+            
+        elif hasattr(delta, "tool_calls") and delta.tool_calls:
+            for tool_call in delta.tool_calls:
+                if tool_call.function is not None:
+                    function = tool_call.function
+                    yield FunctionFragment(
+                        name =(
+                            function.name
+                            if hasattr(function, "name") and function.name
+                            else ""
+                        ),
+                        arguments=(
+                            function.arguments
+                            if hasattr(function, "arguments") and function.arguments
+                            else ""
+                        )
+                    )
 
 
 def find_last_punctuation(buffer: str) -> Optional[int]:

vocode/streaming/agent/utils.py

@@ -0,0 +1,47 @@
+import logging
+import torch
+from importlib import resources as impresources
+
+
+class SileroVAD:
+    INT16_NORM_CONST = 32768.0
+
+    def __init__(self, sample_rate: int, window_size: int, threshold: float = 0.5):
+        # Silero VAD is optimized for performance on single CPU thread
+        torch.set_num_threads(1)
+
+        self.logger = logging.getLogger(__name__)
+        self.model = self._load_model(use_onnx=False)
+        self.sample_rate = sample_rate
+        self.threshold = threshold
+        self.window_size = window_size
+
+    def _load_model(self, use_onnx: bool = False) -> torch.nn.Module:
+        try:
+            model, _ = torch.hub.load(
+                repo_or_dir='silero-vad',
+                model='silero_vad',
+                source='local',
+                onnx=use_onnx
+            )
+        except FileNotFoundError:
+            self.logger.warning("Could not find local VAD model, downloading from GitHub!")
+            model, _ = torch.hub.load(
+                repo_or_dir='snakers4/silero-vad',
+                model='silero_vad',
+                source='github',
+                onnx=use_onnx
+            )
+        return model
+
+    def process_chunk(self, chunk: bytes) -> bool:
+        if len(chunk) != self.window_size:
+            raise ValueError(f"Chunk size must be {self.window_size} bytes")
+        chunk_array = torch.frombuffer(chunk, dtype=torch.int16).to(torch.float32) / self.INT16_NORM_CONST
+        speech_prob = self.model(chunk_array, self.sample_rate).item()
+        if speech_prob > self.threshold:
+            return True
+        return False
+
+    def reset_states(self) -> None:
+        self.model.reset_states()