model_id = "huggingface-asr-whisper-large-v3"# Deploying the modelfrom sagemaker.jumpstart.model import JumpStartModelfrom sagemaker.serializers import JSONSerializermy_model = JumpStartModel(model_id=model_id,instance_type='ml.g5.2xlarge')predictor = my_model.deploy()

max_length：模型生成文本直到输出长度达到该值。language 和 task：我们在这里指定输出语言和任务。transcribe 或者 translate。max_new_tokens：生成的最大 token 数量。num_return_sequences：返回的输出序列数量。num_beams：在贪婪搜索中使用的 beam 数量。如果指定，它必须是一个大于或等于`num_return_sequences`的整数。no_repeat_ngram_size：模型确保在输出序列中不会重复`no_repeat_ngram_size`长度的单词序列。temperature：控制输出的随机性。温度越高，输出序列中低概率单词的数量越多；温度越低，输出序列中高概率单词的数量越多。如果`temperature` -> 0，它会导致贪婪解码。early_stopping：如果为 True，当所有 beam 假设都到达句子结束标记时，文本生成就会结束。do_sample：如果为 True，根据概率对下一个单词进行采样。top_k：在文本生成的每一步，只从`top_k`个最有可能的单词中采样。top_p：在文本生成的每一步，从具有累积概率`top_p`的最小可能单词集合中采样。0 到 1 之间的浮点数。

input_audio_file_name = "sample_french1.wav"with open(input_audio_file_name, "rb") as file:  wav_file_read = file.read()payload = {"audio_input": wav_file_read.hex(), "language": "french", "task": "translate"}predictor.serializer = JSONSerializer()predictor.content_type = "application/json"response = predictor.predict(payload)# We will get the output translated to english for the french audio fileprint(response["text"])

def stream_conversation(bedrock_client,          model_id,          messages,          system_prompts,          inference_config,          additional_model_fields,          tool_config,          round=0):  """  Sends messages to a model and streams the response.  Args:    bedrock_client: The Boto3 Bedrock runtime client.    model_id (str): The model ID to use.    messages (JSON) : The messages to send.    system_prompts (JSON) : The system prompts to send.    inference_config (JSON) : The inference configuration to use.    additional_model_fields (JSON) : Additional model fields to use.  Returns:    Nothing.  """  print(f'n************ ROUND {round} START ************')  print("Streaming messages with model %s" % model_id)  bedrock_params = {    "modelId": model_id,    "messages": messages,    "inferenceConfig": inference_config,    "additionalModelRequestFields": additional_model_fields,    "toolConfig": tool_config  }  system = [item for item in system_prompts if item.get('text')]  if system:    bedrock_params['system'] = system  response = bedrock_client.converse_stream( **bedrock_params )  stream = response.get('stream')  tools_buf = {}  resp_text_buf = ''  if stream:    stop_reason = None    metadata = None    for event in stream:      ...省略                  model_id = "anthropic.claude-3-sonnet-20240229-v1:0"  system_prompt = ""# Message to send to the model.input_text = user_queryprint(colored(f"Question: {input_text}", 'red'))message = {    "role": "user",    "content": [{"text": input_text}]}messages = [message]  # System prompts.system_prompts = [{"text" : system_prompt}]# inference parameters to use.temperature = 0.9top_k = 200max_tokens = 2000# Base inference parameters.inference_config = {    "temperature": temperature,    "maxTokens": max_tokens, }# Additional model inference parameters.additional_model_fields = {"top_k": top_k}bedrock_client = boto3.client(service_name='bedrock-runtime')stream_conversation(bedrock_client, model_id, messages,            system_prompts, inference_config, additional_model_fields, tool_config)

[{"role":"日语翻译","instruct":"你是一个专业的日语翻译，注意翻译时需要考虑专业词汇，请把以下翻译成日语:"},{"role":"图文问答","instruct":"先描述一下该图像的内容，再回答问题"},{"role":"知识问答","instruct":"根据企业知识文档回答问题，如果没有找到具体的文档，请按你掌握的知识尽全力回答。我的问题是"},{"role":"职业顾问","instruct":"我想让你担任职业顾问。我将为您提供一个在职业生涯中寻求指导的人，您的任务是帮助他们根据自己的技能、兴趣和经验确定最适>合的职业。您还应该对可用的各种选项进行研究，解释不同行业的就业市场趋势，并就哪些资格对追求特定领域有益提出建议。我的第一个请求是"},{"role":"心灵导师","instruct":"从现在起你是一个充满哲学思维的心灵导师，当我每次输入一个疑问时你需要用一句富有哲理的名言警句来回答我，并且表明作者和出处要求字数不少于15个字，每次只返回一句且不输出额外的其他信息，你需要使用中文输出"}]

def initial_role_prompt(role_template_path:str):  global role_keys,role_values,role_prompt_dict  with open(role_template_path) as f:    json_data = json.load(f)    for item in json_data:      role_keys.append(item["role"])      role_values.append(item["instruct"])      role_prompt_dict[item["role"]]=item["instruct"]if __name__ == '__main__':  initial_role_prompt("./role_template.json")  ...省略  with gr.Row():    dropdown = gr.Dropdown(role_prompt_dict,value="")    system_prompts = gr.Textbox(role_values[0],visible=False)  ...省略

refer_wav_path":默认参考音频路径"prompt_text":推理TTS的文本"prompt_language":"ja","text":" 默认参考音频文本"text_language" :`默认参考音频语种, "中文","英文","日文","zh","en","ja"`,"cut_punc":"文本切分符号设定, 默认为空, 以",.，。"字符串的方式传入

def pre_download(ref_wav_path:str)-> None:  if "s3" in ref_wav_path:    file_name = os.path.basename(ref_wav_path)    download_file = "/tmp/"+file_name    download_from_s3(ref_wav_path,download_file)    return download_file  else:    return ref_wav_path

"""uvicorn main module"""import timeimport osimport subprocessimport signalimport uvicornimport sys#  ###for debug only#######sys.path.append(os.path.join(os.path.dirname(__file__), "lib"))import sagemaker_ssh_helpersagemaker_ssh_helper.setup_and_start_ssh()def _add_sigterm_handler(mms_process):  def _terminate(signo, frame): # pylint: disable=unused-argument    try:      os.system('ps aux')      os.kill(mms_process.pid, signal.SIGTERM)    except OSError:      pass  signal.signal(signal.SIGTERM, _terminate)cmd = ["python","/opt/program/api.py","-dr","Brigida.wav","-dt","hi hello","-dl","zh","-a","0.0.0.0","-p","8080"]process = subprocess.Popen(cmd)process.wait()

RUN pip install sagemaker-ssh-helperRUN pip install boto3RUN pip3 install pydantic#RUN curl -L https://github.com/peak/s5cmd/releases/download/v2.2.2/s5cmd_2.2.2_Linux-64bit.tar.gz | tar -xz && mv s5cmd /opt/program/ENV PYTHONUNBUFFERED=TRUEENV PYTHONDONTWRITEBYTECODE=TRUEENV PATH="/opt/program:${PATH}"# Install 3rd party appsENV DEBIAN_FRONTEND=noninteractiveENV TZ=Etc/UTCRUN apt-get update &&   apt-get install -y --no-install-recommends tzdata ffmpeg libsox-dev parallel aria2 git git-lfs &&   git lfs install &&   rm -rf /var/lib/apt/lists/*# Copy to leverage Docker cacheCOPY ./GPT-SoVITS /opt/program/WORKDIR /opt/programRUN pip install --no-cache-dir -r /opt/program/requirements.txt# Define a build-time argument for image typeARG IMAGE_TYPE=full# Conditional logic based on the IMAGE_TYPE argument# Always copy the Docker directory, but only use it if IMAGE_TYPE is not "elite"COPY ./Docker /workspace/Docker#####start api.py ######RUN chmod 755 /opt/programRUN chmod 755 /opt/program/serve

request = {"refer_wav_path":"s3://sagemaker-us-west-2-687912291502/gpt-sovits/wav/123.WAV.wav",  "prompt_text": "早上好，欢迎来到我的一天，这是我做音频主播的第四个年头了，有小伙伴留言想让我分享一些音频直播的经验，今天我先和大家聊聊我的入行原因。其实我从中选情就很喜欢电台了，在校期间眼睛记得参加各类广播站以及校园主持的活动，我好像对话筒和声音就有一种莫名的执念，所以在17年里的时候，我创建了一个自己的电台公众号，通过声音和文字记录自己的一些心事。后来18年底经由朋友的介绍，可以通过音频直播分享我写的东西，还有我的声音。当时呢我就想音频直播又不用落脸那么方便，试一试吧，如果有人喜欢，还可以给自己的电台公众号吸吸粉。后来我就在直播间里认识了越来越多的听友，渐渐的这份工作，也为我带来了一些兼职收入，我就决定把这份工作做下去。",  "prompt_language":"zh",  "text":"作为SAP基础架构专家,我来解释一下SAP Basis的含义:SAP Basis是指SAP系统的基础设施层,负责管理和维护整个SAP系统环境的运行。它包括以下几个主要方面:SAP系统管理包括SAP系统实例的安装、启动、监控、备份、升级等日常管理任务。Basis团队负责保证系统的正常运行。",  "text_language" :"zh",  "output_s3uri":"s3://sagemaker-us-west-2-687912291502/gpt_sovits_output/wav/"}def invoke_endpoint():  content_type = "application/json"  request_body = request  payload = json.dumps(request_body)  print(payload)  response = runtime_sm_client.invoke_endpoint(    EndpointName=endpointName,    ContentType=content_type,    Body=payload,  )  result = response['Body'].read().decode()  print('返回：',result)

class InferenceOpt(BaseModel):  refer_wav_path: str = "",  prompt_text: str = "",  prompt_language:str = "zh",  text:str = "my queue, my love ,my wife.",  text_language :str = "zh"  output_s3uri:str = "s3://sagemaker-us-west-2-687912291502/gpt_sovits_output/wav/"  cut_punc:str = "."app = FastAPI()@app.get("/ping")async def ping():  """  ping /ping func  """  return {"message": "ok"}@app.post("/invocations")async def invocations(request: Request):  json_post_raw = await request.json()  print(f"invocations {json_post_raw=}")  opt=parse_obj_as(InferenceOpt,json_post_raw)  print(f"invocations {opt=}")  return handle(opt.refer_wav_path, opt.prompt_text, opt.prompt_language, opt.text, opt.text_language,opt.cut_punc, opt.output_s3uri)

## 根据分隔符切割生成文本if cut_punc == None:    texts = cut_text(text,default_cut_punc)  else:    texts = cut_text(text,cut_punc)for text in texts:  ## 每段chunk TTS推理合成语音  ...省略  if stream_mode == "normal":    audio_bytes, chunked_audio_bytes = read_clean_buffer(audio_bytes)    ## 打包每段chunk推理语音并put到s3    chunked_audio_bytes = pack_mp3(chunked_audio_bytes,hps.data.sampling_rat    result = write_wav_to_s3(chunked_audio_bytes,output_s3uri)    ## yield生成器返回，以便SageMaker endpoint 流式输出    yield json.dumps(result)

def invoke_streams_endpoint(smr_client,endpointName, request):  global chunk_bytes  content_type = "application/json"  payload = json.dumps(request,ensure_ascii=False)  response_model = smr_client.invoke_endpoint_with_response_stream(    EndpointName=endpointName,    ContentType=content_type,    Body=payload,  )  result = []  print(response_model['ResponseMetadata'])  event_stream = iter(response_model['Body'])  index = 0  try:     while True:      event = next(event_stream)      eventChunk = event['PayloadPart']['Bytes']      chunk_dict = {}      if index == 0:        print("Received first chunk")        chunk_dict['first_chunk'] = True        chunk_dict['bytes'] = eventChunk        chunk_bytes = eventChunk        chunk_dict['last_chunk'] = False        chunk_dict['index'] = index      else:        chunk_dict['first_chunk'] = False        chunk_dict['bytes'] = eventChunk        chunk_bytes = eventChunk        chunk_dict['last_chunk'] = False        chunk_dict['index'] = index      print("chunk len:",len(chunk_dict['bytes']))      result.append(chunk_dict)        index += 1      #print('返回chunk：', chunk_dict['bytes'])  except StopIteration:    print("All chunks processed")    chunk_dict = {}    chunk_dict['first_chunk'] = False    chunk_dict['bytes'] = chunk_bytes    chunk_dict['last_chunk'] = True    chunk_dict['index'] = index-1    result = upsert(result,chunk_dict)  print("result",result)  return result