jandres/GenAIExamples
1
0
Fork 0
Code Issues Pull Requests Actions 5 Packages Projects Releases Wiki Activity

Adds audio querying to MultimodalQ&A Example (#1225)

Browse Source 
Signed-off-by: Melanie Buehler <melanie.h.buehler@intel.com>
Signed-off-by: okhleif-IL <omar.khleif@intel.com>
Signed-off-by: dmsuehir <dina.s.jones@intel.com>
Co-authored-by: Omar Khleif <omar.khleif@intel.com>
Co-authored-by: Dina Suehiro Jones <dina.s.jones@intel.com>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Abolfazl Shahbazi <12436063+ashahba@users.noreply.github.com>
This commit is contained in:
Melanie Hart Buehler
2024-12-12 00:05:14 -08:00
committed by GitHub
parent a50e4e6f9f
commit c760cac2f4
13 changed files with 389 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

147
MultimodalQnA/multimodalqna.py
View File

@@ -2,6 +2,7 @@
# SPDX-License-Identifier: Apache-2.0
import base64
import json
import os
from io import BytesIO
@@ -16,7 +17,7 @@ from comps.cores.proto.api_protocol import (
)
from comps.cores.proto.docarray import LLMParams
from fastapi import Request
from fastapi.responses import StreamingResponse
from fastapi.responses import JSONResponse, StreamingResponse
from PIL import Image
MEGA_SERVICE_PORT = int(os.getenv("MEGA_SERVICE_PORT", 8888))
@@ -29,6 +30,9 @@ LVM_SERVICE_PORT = int(os.getenv("LVM_SERVICE_PORT", 9399))
class MultimodalQnAService(Gateway):
asr_port = int(os.getenv("ASR_SERVICE_PORT", 3001))
asr_endpoint = os.getenv("ASR_SERVICE_ENDPOINT", "http://0.0.0.0:{}/v1/audio/transcriptions".format(asr_port))
def __init__(self, host="0.0.0.0", port=8000):
self.host = host
self.port = port
@@ -73,7 +77,10 @@ class MultimodalQnAService(Gateway):
# this overrides _handle_message method of Gateway
def _handle_message(self, messages):
images = []
audios = []
b64_types = {}
messages_dicts = []
decoded_audio_input = ""
if isinstance(messages, str):
prompt = messages
else:
@@ -87,16 +94,26 @@ class MultimodalQnAService(Gateway):
system_prompt = message["content"]
elif msg_role == "user":
if type(message["content"]) == list:
# separate each media type and store accordingly
text = ""
text_list = [item["text"] for item in message["content"] if item["type"] == "text"]
text += "\n".join(text_list)
image_list = [
item["image_url"]["url"] for item in message["content"] if item["type"] == "image_url"
]
if image_list:
messages_dict[msg_role] = (text, image_list)
else:
audios = [item["audio"] for item in message["content"] if item["type"] == "audio"]
if audios:
# translate audio to text. From this point forward, audio is treated like text
decoded_audio_input = self.convert_audio_to_text(audios)
b64_types["audio"] = decoded_audio_input
if text and not audios and not image_list:
messages_dict[msg_role] = text
elif audios and not text and not image_list:
messages_dict[msg_role] = decoded_audio_input
else:
messages_dict[msg_role] = (text, decoded_audio_input, image_list)
else:
messages_dict[msg_role] = message["content"]
messages_dicts.append(messages_dict)
@@ -108,55 +125,84 @@ class MultimodalQnAService(Gateway):
if system_prompt:
prompt = system_prompt + "\n"
for messages_dict in messages_dicts:
for i, (role, message) in enumerate(messages_dict.items()):
for i, messages_dict in enumerate(messages_dicts):
for role, message in messages_dict.items():
if isinstance(message, tuple):
text, image_list = message
text, decoded_audio_input, image_list = message
if i == 0:
# do not add role for the very first message.
# this will be added by llava_server
if text:
prompt += text + "\n"
elif decoded_audio_input:
prompt += decoded_audio_input + "\n"
else:
if text:
prompt += role.upper() + ": " + text + "\n"
elif decoded_audio_input:
prompt += role.upper() + ": " + decoded_audio_input + "\n"
else:
prompt += role.upper() + ":"
for img in image_list:
# URL
if img.startswith("http://") or img.startswith("https://"):
response = requests.get(img)
image = Image.open(BytesIO(response.content)).convert("RGBA")
image_bytes = BytesIO()
image.save(image_bytes, format="PNG")
img_b64_str = base64.b64encode(image_bytes.getvalue()).decode()
# Local Path
elif os.path.exists(img):
image = Image.open(img).convert("RGBA")
image_bytes = BytesIO()
image.save(image_bytes, format="PNG")
img_b64_str = base64.b64encode(image_bytes.getvalue()).decode()
# Bytes
else:
img_b64_str = img
images.append(img_b64_str)
else:
if image_list:
for img in image_list:
# URL
if img.startswith("http://") or img.startswith("https://"):
response = requests.get(img)
image = Image.open(BytesIO(response.content)).convert("RGBA")
image_bytes = BytesIO()
image.save(image_bytes, format="PNG")
img_b64_str = base64.b64encode(image_bytes.getvalue()).decode()
# Local Path
elif os.path.exists(img):
image = Image.open(img).convert("RGBA")
image_bytes = BytesIO()
image.save(image_bytes, format="PNG")
img_b64_str = base64.b64encode(image_bytes.getvalue()).decode()
# Bytes
else:
img_b64_str = img
images.append(img_b64_str)
elif isinstance(message, str):
if i == 0:
# do not add role for the very first message.
# this will be added by llava_server
if message:
prompt += role.upper() + ": " + message + "\n"
prompt += message + "\n"
else:
if message:
prompt += role.upper() + ": " + message + "\n"
else:
prompt += role.upper() + ":"
if images:
return prompt, images
b64_types["image"] = images
# If the query has multiple media types, return all types
if prompt and b64_types:
return prompt, b64_types
else:
return prompt
def convert_audio_to_text(self, audio):
# translate audio to text by passing in base64 encoded audio to ASR
if isinstance(audio, dict):
input_dict = {"byte_str": audio["audio"][0]}
else:
input_dict = {"byte_str": audio[0]}
response = requests.post(self.asr_endpoint, data=json.dumps(input_dict))
if response.status_code != 200:
return JSONResponse(
status_code=503, content={"message": "Unable to convert audio to text. {}".format(response.text)}
)
response = response.json()
return response["query"]
async def handle_request(self, request: Request):
data = await request.json()
stream_opt = bool(data.get("stream", False))
@@ -165,16 +211,35 @@ class MultimodalQnAService(Gateway):
stream_opt = False
chat_request = ChatCompletionRequest.model_validate(data)
# Multimodal RAG QnA With Videos has not yet accepts image as input during QnA.
prompt_and_image = self._handle_message(chat_request.messages)
if isinstance(prompt_and_image, tuple):
# print(f"This request include image, thus it is a follow-up query. Using lvm megaservice")
prompt, images = prompt_and_image
num_messages = len(data["messages"]) if isinstance(data["messages"], list) else 1
messages = self._handle_message(chat_request.messages)
decoded_audio_input = ""
if num_messages > 1:
# This is a follow up query, go to LVM
cur_megaservice = self.lvm_megaservice
initial_inputs = {"prompt": prompt, "image": images[0]}
if isinstance(messages, tuple):
prompt, b64_types = messages
if "audio" in b64_types:
# for metadata storage purposes
decoded_audio_input = b64_types["audio"]
if "image" in b64_types:
initial_inputs = {"prompt": prompt, "image": b64_types["image"][0]}
else:
initial_inputs = {"prompt": prompt, "image": ""}
else:
prompt = messages
initial_inputs = {"prompt": prompt, "image": ""}
else:
# print(f"This is the first query, requiring multimodal retrieval. Using multimodal rag megaservice")
prompt = prompt_and_image
# This is the first query. Ignore image input
cur_megaservice = self.megaservice
if isinstance(messages, tuple):
prompt, b64_types = messages
if "audio" in b64_types:
# for metadata storage purposes
decoded_audio_input = b64_types["audio"]
else:
prompt = messages
initial_inputs = {"text": prompt}
parameters = LLMParams(
@@ -207,18 +272,24 @@ class MultimodalQnAService(Gateway):
if "text" in result_dict[last_node].keys():
response = result_dict[last_node]["text"]
else:
# text in not response message
# text is not in response message
# something wrong, for example due to empty retrieval results
if "detail" in result_dict[last_node].keys():
response = result_dict[last_node]["detail"]
else:
response = "The server fail to generate answer to your query!"
response = "The server failed to generate an answer to your query!"
if "metadata" in result_dict[last_node].keys():
# from retrieval results
metadata = result_dict[last_node]["metadata"]
if decoded_audio_input:
metadata["audio"] = decoded_audio_input
else:
# follow-up question, no retrieval
metadata = None
if decoded_audio_input:
metadata = {"audio": decoded_audio_input}
else:
metadata = None
choices = []
usage = UsageInfo()
choices.append(