Spaces:
Sleeping
Sleeping
File size: 2,967 Bytes
bd435b3 245e7f9 bd435b3 245e7f9 bd435b3 245e7f9 bd435b3 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 |
import json
import os
import torch
import openai
from audio_stream_processor import AudioStreamProcessor
from speech_service import SpeechService
class StreamingChatService:
def __init__(self, audio_processor:AudioStreamProcessor()=None, api="openai", model_id = "gpt-3.5-turbo", voice_id="Bella"):
self._audio_processor = audio_processor
self._speech_service = SpeechService(voice_id=voice_id)
self._api = api
self._device = "cuda:0" if torch.cuda.is_available() else "cpu"
self._system_prompt = None
openai.api_key = os.getenv("OPENAI_API_KEY")
self._model_id = model_id
self.reset()
def reset(self):
self._messages = []
if self._system_prompt:
self._messages.append({"role": "system", "content": self._system_prompt})
def _should_we_send_to_voice(self, sentence):
sentence_termination_characters = [".", "?", "!"]
temination_charicter_present = any(c in sentence for c in sentence_termination_characters)
if temination_charicter_present and sentence[-1] not in sentence_termination_characters:
# text_to_speak = sentence up until the last sentence termination character
termination_indices = [sentence.rfind(char) for char in sentence_termination_characters]
last_termination_index = max(termination_indices)
text_to_speak = sentence[:last_termination_index+1]
return text_to_speak
if temination_charicter_present:
return False
return False
def respond_to(self, prompt):
self._messages.append({"role": "user", "content": prompt})
agent_response = ""
current_sentence = ""
response = openai.ChatCompletion.create(
model=self._model_id,
messages=self._messages,
temperature=1.0, # use 1.0 for debugging/deteministic results
stream=True
)
for chunk in response:
chunk_message = chunk['choices'][0]['delta']
if 'content' in chunk_message:
chunk_text = chunk_message['content']
# print(chunk_text)
current_sentence += chunk_text
agent_response += chunk_text
text_to_speak = self._should_we_send_to_voice(current_sentence)
if text_to_speak:
stream = self._speech_service.stream(text_to_speak)
self._audio_processor.add_audio_stream(stream)
print(text_to_speak)
current_sentence = current_sentence[len(text_to_speak):]
if len(current_sentence) > 0:
stream = self._speech_service.stream(current_sentence)
self._audio_processor.add_audio_stream(stream)
print(current_sentence)
self._messages.append({"role": "assistant", "content": agent_response})
return agent_response
|