import uvicorn from fastapi import FastAPI, UploadFile, File, Form, HTTPException from typing import List, Dict, Any, Optional import json import os from transformers import AutoProcessor, AutoModel import torch, torchaudio import os import copy from rapidfuzz import process, fuzz from pypinyin import pinyin, Style def correct_sentence_with_pinyin(user_input_sentence, location_dict, score_cutoff=50): pinyin_dict = {} for location in location_dict: pinyin_name = ''.join([item[0] for item in pinyin(location, style=Style.NORMAL)]) pinyin_dict[pinyin_name] = location user_pinyin_sentence = ''.join([item[0] for item in pinyin(user_input_sentence, style=Style.NORMAL)]) best_match_pinyin = process.extractOne( query=user_pinyin_sentence, choices=list(pinyin_dict.keys()), # 傳入拼音作為搜尋目標 scorer=fuzz.token_set_ratio, score_cutoff=score_cutoff ) if best_match_pinyin: best_pinyin_name = best_match_pinyin[0] corrected_location_name = pinyin_dict[best_pinyin_name] best_user_substring = None max_substring_score = 0 for i in range(len(user_input_sentence)): for j in range(i + 2, min(i + 16, len(user_input_sentence) + 1)): substring = user_input_sentence[i:j] score = fuzz.ratio(substring, corrected_location_name) if score > max_substring_score: max_substring_score = score best_user_substring = substring if best_user_substring and max_substring_score > score_cutoff: return user_input_sentence.replace(best_user_substring, corrected_location_name, 1) else: return user_input_sentence return user_input_sentence class InferenceClass: def __init__(self,model_id): self.model = AutoModel.from_pretrained( model_id, device_map="cuda", torch_dtype=torch.bfloat16, trust_remote_code=True, attn_implementation="eager" ).eval() self.processor = AutoProcessor.from_pretrained( model_id, trust_remote_code=True ) self.remove_words_signs = lambda x:x.replace('User transcribe is :','').replace('GPT output is :','').replace('\n','').\ replace(' ','').replace('?','').replace('？','').replace('!','').replace('。','').\ replace('！','') def call_gpt(self,inputs_tensor): with torch.inference_mode(): inputs = {k:inputs_tensor[k].to('cuda') for k in inputs_tensor} generate_ids = self.model.generate(**inputs, max_new_tokens=128, do_sample=False) generate_ids = generate_ids[:, inputs['input_ids'].shape[1] :] model_output = self.processor.batch_decode( generate_ids, skip_special_tokens=True, clean_up_tokenization_spaces=False )[0] return model_output def call_function_fake(self,messages=[],obs=""): messages.append({'from': 'observation', 'value': obs}) return messages def generate(self,chat_history,tools="",audio_path=None): ''' input: audio_path : str chat_history : dict return: model_output : dict ''' chat_history = copy.deepcopy(chat_history) if type(audio_path)!=type(None): chat_history.append({'from': 'human', 'value': [{'type': 'audio', 'audio': audio_path}]}) words_from_poi = [] for hist in chat_history: if hist['from']=='observation' and '地點查詢成功' in hist['value'] and 'poi' in hist['value']: tmp = json.loads(hist['value']) for i,poi in enumerate(tmp['poi']): words_from_poi.append(poi['name']) for hist in chat_history: if hist['from']=='human' and type(hist['value'])==str: hist['value'] = correct_sentence_with_pinyin(hist['value'],words_from_poi) elif hist['from']=='function_call' and "arguments" in hist['value'] and 'keyword' in hist['value']["arguments"]: hist['value']["arguments"] = eval(hist['value']["arguments"]) if 'keyword' in hist['value']["arguments"]: hist['value']["arguments"]['keyword'] = correct_sentence_with_pinyin(hist['value']["arguments"]['keyword'],words_from_poi) hist['value']["arguments"] = str(hist['value']["arguments"]) # model_input_history = copy.deepcopy(chat_history) # num2ch = {1:'一',2:'二',3:'三',4:'四',5:'五',6:'六'} # for hist in model_input_history: # if hist['from']=='observation' and '地點查詢成功' in hist['value'] and 'poi' in hist['value']: # tmp = json.loads(hist['value']) # new_poi = [] # for i,poi in enumerate(tmp['poi']): # new_poi.append('第{}個 : '.format(num2ch[i+1])+str(poi)) # tmp['poi'] = new_poi # hist['value'] = json.dumps(tmp, ensure_ascii=False) inputs_text = self.processor.apply_chat_template( chat_history, add_generation_prompt=True, tokenize=False, return_dict=True, return_tensors="pt", tools=json.loads(tools) ) inputs_tensor = self.processor(text=inputs_text, audio=[torchaudio.load(audio_path)[0]] if type(audio_path)!=type(None) else None, add_special_tokens=False, return_tensors='pt' ) model_output = self.call_gpt(inputs_tensor) if chat_history[-1]['from']=='observation': chat_history.append({'from': 'gpt', 'value': correct_sentence_with_pinyin(model_output,words_from_poi)}) return chat_history if ((not ';\n' in model_output) or (not 'User transcribe is :' in model_output) or (not 'GPT output is :' in model_output)\ or len(model_output.split(';\n'))<2 ): if chat_history[-1]['value']!="抱歉我聽不清楚 能麻煩您再說一次嗎": chat_history.append({'from': 'human', 'value': 'HUMAN_VOICE_IS_NOT_RECOGNIZED'}), chat_history.append({'from': 'gpt', 'value': '抱歉我聽不清楚 能麻煩您再說一次嗎'}) return chat_history output_t,output_o = model_output.split(';\n')[:2] output_t,output_o = self.remove_words_signs(output_t),self.remove_words_signs(output_o) chat_history[-1]['value'] = correct_sentence_with_pinyin(output_t,words_from_poi) if 'Action:' in output_o and 'ActionInput:' in output_o: # function calling function_name,function_arg = output_o.split('ActionInput:') function_name = function_name.replace('Action:','') if "keyword" in function_arg: function_arg = json.loads(function_arg) if "keyword" in function_arg: function_arg["keyword"] = correct_sentence_with_pinyin(function_arg["keyword"],words_from_poi) chat_history.append({'from': 'function_call', 'value': {"name": function_name, "arguments": str(function_arg)}}) else: # gpt response chat_history.append({'from': 'gpt', 'value': correct_sentence_with_pinyin(output_o,words_from_poi)}) return chat_history model_id = "/home/jeff/jeff/codes/llm/InCar/gemma-3-4b-it-omni" pipeline = InferenceClass(model_id) app = FastAPI( title="Audio LLM API", description="An API that accepts an audio file and a list of dictionaries.", ) import json dataset = json.load(open('/home/jeff/jeff/codes/llm/InCar/data/test_data/nav_0730_noisy.json')) tools = dataset[0]['tools'] @app.post("/audio_llm/") async def process_audio_and_data( audio_file: Optional[UploadFile] = File(None, description="The audio file to be processed."), data: str = Form(..., description="A JSON string representing a list of chat history dictionaries.") ) -> List[Dict[str, Any]]: try: input_data_list = json.loads(data) if not isinstance(input_data_list, list) or not all(isinstance(item, dict) for item in input_data_list): raise ValueError("The provided data is not a list of dictionaries.") except json.JSONDecodeError: raise HTTPException( status_code=422, detail="Invalid JSON format for 'data' field. Please provide a valid JSON string." ) except ValueError as e: raise HTTPException( status_code=422, detail=str(e) ) temp_file_path=None if audio_file: temp_file_path = f"./audio_path/temp_{audio_file.filename}" with open(temp_file_path, "wb") as buffer: buffer.write(await audio_file.read()) print(f"Audio file saved to {temp_file_path}") output_data = pipeline.generate(input_data_list,tools=tools,audio_path=temp_file_path) print(output_data) return output_data # uvicorn main:app --host 0.0.0.0 --port 8087 --log-level info --workers 1 >> ./log.txt if __name__ == "__main__": uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)