Spaces:
Sleeping
Sleeping
File size: 3,740 Bytes
5a813b2 5876c46 |
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 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 |
import base64
import cv2
import openai
import streamlit as st
from dotenv import load_dotenv
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.schema.messages import SystemMessage
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI
from pyaudio import PyAudio, paInt16
from speech_recognition import Microphone, Recognizer, UnknownValueError
load_dotenv()
class WebcamStream:
def __init__(self):
self.stream = cv2.VideoCapture(index=0)
_, self.frame = self.stream.read()
self.running = False
def start(self):
self.running = True
return self
def update(self):
while self.running:
_, frame = self.stream.read()
_, buffer = cv2.imencode(".jpeg", frame)
self.frame = base64.b64encode(buffer).decode()
def read(self):
return self.frame
def stop(self):
self.running = False
self.stream.release()
class Assistant:
def __init__(self, model):
self.chain = self._create_inference_chain(model)
def answer(self, prompt, image):
if not prompt:
return
print("Prompt:", prompt)
response = self.chain.invoke(
{"prompt": prompt, "image_base64": image},
config={"configurable": {"session_id": "unused"}},
).strip()
print("Response:", response)
if response:
self._tts(response)
def _tts(self, response):
player = PyAudio().open(format=paInt16, channels=1, rate=24000, output=True)
with openai.audio.speech.with_streaming_response.create(
model="tts-1",
voice="alloy",
response_format="pcm",
input=response,
) as stream:
for chunk in stream.iter_bytes(chunk_size=1024):
player.write(chunk)
def _create_inference_chain(self, model):
SYSTEM_PROMPT = """
You are a witty assistant that will use the chat history and the image
provided by the user to answer its questions. Your job is to answer
questions.
Use few words on your answers. Go straight to the point. Do not use any
emoticons or emojis.
Be friendly and helpful. Show some personality.
"""
prompt_template = ChatPromptTemplate.from_messages(
[
SystemMessage(content=SYSTEM_PROMPT),
MessagesPlaceholder(variable_name="chat_history"),
(
"human",
[
{"type": "text", "text": "{prompt}"},
{
"type": "image_url",
"image_url": "data:image/jpeg;base64,{image_base64}",
},
],
),
]
)
chain = prompt_template | model | StrOutputParser()
chat_message_history = ChatMessageHistory()
return RunnableWithMessageHistory(
chain,
lambda _: chat_message_history,
input_messages_key="prompt",
history_messages_key="chat_history",
)
def audio_callback(recognizer, audio):
try:
prompt = recognizer.recognize_whisper(audio, model="base", language="english")
assistant.answer(prompt, webcam_stream.read())
except UnknownValueError:
print("There was an error processing the audio.")
def main():
st.title("AI Assistant")
webcam_stream = WebcamStream() |