#🔒 Looking for a multithreading/multiprocessing Wizard
prisma thorn
TLDR: Trying to make a Discord bot using python. The bot uses Ollama (OpenAI API) to generate responses and ASR for audio transcription.
My problem; I can't figure out how to make it transcribe the audio while still listening for incoming.
All the code is on github: https://github.com/AnthonMS/DiscOllama
The code has become extremely messy, so dont judge me too much. I'm way out of my comfort zone. Both in terms of difficulty in regards to python, but also, haven't done much multithreading/multiprocessing at all in any language. It will probably be a lengthy process to get to a proper solution.
Its the VoiceChat.py I am working with
When joining a voicechat, it starts listening with the listen function
def listen(self, user, data: voice_recv.VoiceData):
# ...
if time_speaking >= timedelta(milliseconds=500):
audio_data = self.user_audio[user.id]['audio']
processed_data = self.user_audio[user.id]['processed_audio']
new_audio_data = audio_data[len(processed_data):]
if is_silence(new_audio_data[-6400:], 5): # Past 200ms is very silent
# ...
## Maybe instead of trying to start a transcription task or do it from here. We add it to a queue and then have another task checking for queued audio_data?
# loop_handler.run_coroutine(self.transcribe_user_audio(new_audio_data, filename, user.id)) # This runs it as a loop. I just need to run it once. AI Made it lol.
# self.transcribe_user_audio(new_audio_data, filename, user.id) # This blocks it from listening until it's done transcribing
# # self.test_task = asyncio.create_task(self.transcribe_user_audio(new_audio_data, filename, user.id)) ## Error: No running event loop
async def transcribe_user_audio(self, audio_data, filename, user_id):
self.active_transcriptions += 1
textResult = None
try:
asr_data = bytes_to_float32_array(audio_data)
start_time = datetime.now()
text = self.asr(asr_data)
textResult = text['text']
end_time = datetime.now()
elapsed_time = end_time - start_time
except asyncio.CancelledError:
pass
except Exception as e:
logging.error("Error transcribing audio: ")
logging.error(e)
finally:
self.active_transcriptions -= 1
if (textResult == None):
return
self.new_messages = True
logging.info(f"Result: {textResult}, Time taken: {elapsed_time.total_seconds():.2f} seconds")
# Replace filename string with text result
for i, str in enumerate(self.user_audio[user_id]['text']):
if str == filename:
self.user_audio[user_id]['text'][i] = textResult.strip()
break
async def transcribe_user_audio(self, audio_data, filename, user_id):
self.active_transcriptions += 1
textResult = None
try:
asr_data = bytes_to_float32_array(audio_data)
start_time = datetime.now()
text = self.asr(asr_data)
textResult = text['text']
end_time = datetime.now()
elapsed_time = end_time - start_time
except asyncio.CancelledError:
pass
except Exception as e:
logging.error("Error transcribing audio: ")
logging.error(e)
finally:
self.active_transcriptions -= 1
if (textResult == None):
return
self.new_messages = True
logging.info(f"Result: {textResult}, Time taken: {elapsed_time.total_seconds():.2f} seconds")
# Replace filename string with text result
for i, str in enumerate(self.user_audio[user_id]['text']):
if str == filename:
self.user_audio[user_id]['text'][i] = textResult.strip()
break
I have tried using asyncio in many different ways. I have tried creating threads manually. Nothing really seems to work perfect. And that is definetily because I don't really know what I'm doing or how to do it.
I got it to transcribe everything fine if I use the AsyncLoopThread class loop_handler.run_coroutine
But then I cant get it to run the async for loop in the respond function
def on_silence(self):
text_items = []
new_messages = []
for user_id, userdata in self.user_audio.items():
if len(userdata['text']) > 0:
text = ' '.join(userdata['text'])
text_item = {
'user_id': user_id,
'username': userdata['username'],
'text': text,
'time': userdata['last_spoke']
}
text_items.append(text_item)
new_messages.append({
"role": "assistant" if user_id == self.discord.user.id else "user",
"content": text,
})
userdata['text'] = []
userdata['started_speaking'] = None
userdata['processed_audio'] = bytearray()
userdata['audio'] = bytearray()
if (len(text_items) > 0):
text_items.sort(key=lambda x: x['time'])
for item in text_items:
logging.info(f"{item['time']}: User {item['user_id']}: {item['text']}")
self.save_message(item['text'], item['user_id'], item['username'])
if self.active_transcriptions == 0:
# await self.respond(new_messages)
loop_handler.run_coroutine(self.respond(new_messages))
async def respond(self, messages=[]):
try:
saved_messages = self.get_messages()
if (len(saved_messages) == 0):
saved_messages = messages
if (len(saved_messages) == 0):
logging.info("No messages to respond to...")
return
logging.info("Responding to messages: " + str(saved_messages))
current_model = self.get_current_model()
full_response = "" # ollama, model, messages
async for part in self.discOllama.chat(saved_messages, milliseconds=None, model=current_model):
part_content = part['message']['content']
# logging.info(f"Part: {part_content}")
full_response += part_content
logging.info("Full response: " + full_response)
except asyncio.CancelledError:
logging.info("Responding cancelled")
except Exception as e:
logging.error("Error answering")
logging.error(e)
finally:
# logging.info("FINALLY RESPOND: " + full_response)
pass
Again, the code has become extremely messy and I didnt account for the fact that I would need this much multiprocessing. But of course.
The transcription as it is now, works kinda. It can transcribe the audio, and generate a response if no-one else is speaking. But if someone speaks while it is generating this response, then it doesnt "hear" it. The respond is blocking the listen call. I need to make the respond process a new process, and then be able to cancel that process if someone starts speaking again. and the respond function should catch the cancellation so it can save the partial response it has generated.
Roast my code, it's very bad. But add some feedback pointing me in the right direction at least, to make it hurt less. lol. If you wanna jump in a call and brainstorm it out, I welcome you to DM me!
Actually, I mean that the transcription function is blocking the listening. It has been a long day.
The code on github has comments describing what the different lines does. and some thoughts of workarounds on line 100,101,102
https://github.com/AnthonMS/DiscOllama/blob/main/src/VoiceChat.py
livid epoch
uh nuh
prisma thorn
🤣
lone flameBOT
Python help channel closed