import gradio as gr from TritonServerClient import TritonServerClient, InferInput, InferRequestedOutput from TritonServerClient.utils import np_to_triton_dtype from functools import wraps import numpy as np # def prepare_params(query): # bs=1 # # query = "#角色：直播间文案写作专家。#任务：根据用户输入的商品信息写出用于推广营销的直播文案。#目标：引导用户购买直播间商品，提高销量。#要求：- 如果没有输入商品信息，则邀请输入。- 文案写作 - 按照“引出话题”、“产品讲解”、“购买方式”三个方面写作文案。 - 在“产品讲解”部分详细讲解产品细节： - 【重要】充分扩充内容，吸引用户购买。 - 详细描述商品的优惠活动信息（如用户输入没有，不要询问，直接补充）。 - 在“购买方式”部分： - 仅包括引导用户点击链接购买的内容。 - 不要包括快递物流承诺。 - 【重要】每个部分多于5句话。 - 使用口语化的文字。 - 表达方式有亲和力和感染力。 - 对用户们使用“宝子们”、“宝宝们”、“家人们”等有亲和力的称呼。 - 多使用自问自答的语句。- 当你准备好了回复“请输入商品描述和优惠信息”我们就开始。#输出：- 包含在""里。- 写出每一部分内容的标题，包括【引出话题】、【产品讲解】、【购买方式】。- 示例：""【引出话题】【产品讲解】【购买方式】""- 输入商品信息:%s" % prompt # # query = '请你扮演直播间直播卖货的主播角色，根据用户提出的问题，检索知识库中的信息，如果查找到相关的信息，使用口语化的文字，用非常简短的一句话回答用户的问题；如果查找不到相关的信息，直接回答没有，并引导用户关注本场直播商品。你的目的是激发用户的购买欲、提高直播商品销量。回答对用户们使用“宝子”、“宝宝”、“亲”等有亲和力的称呼。\n\n# 用户提出的问题：\n这里面有可乐吗\n\n# 知识库：\n商品名称：外送欢乐桶(券)兑换券\n直播链接序号：64号链接号链接\n商品现价：66元\n商品原价：142元\n折扣：5\n品牌名称：麦当劳麦乐送\n券包张数：1\n套餐详情：["麦辣鸡腿汉堡1个","板烧鸡腿堡1个","中可乐3杯","四拼小食桶A"]\n补充：四拼小食桶A(麦麦脆汁鸡(鸡腿)1块＋中薯条1份＋麦辣鸡翅2块＋麦乐鸡5块)＋麦辣鸡腿堡1个＋板烧鸡腿堡1个＋中可乐3杯 原材料：面包\n\n\n# 输出格式：\n答：\n\n# 要求\n对于用户问到了知识库中未提及的信息不要编造，直接不回答' # ###### # ##### # title_text = np.array([query.encode('utf-8')], dtype=np.string_) # title_text = np.tile(title_text, (bs, 1)) # data_batch = {} # data_batch['query'] = title_text # inputs = [ # InferInput("query", data_batch['query'].shape, # np_to_triton_dtype(data_batch['query'].dtype)), # ] # inputs[0].set_data_from_numpy(data_batch['query']) # return inputs def prepare_params(query,img): bs=1 # query = "#角色：直播间文案写作专家。#任务：根据用户输入的商品信息写出用于推广营销的直播文案。#目标：引导用户购买直播间商品，提高销量。#要求：- 如果没有输入商品信息，则邀请输入。- 文案写作 - 按照“引出话题”、“产品讲解”、“购买方式”三个方面写作文案。 - 在“产品讲解”部分详细讲解产品细节： - 【重要】充分扩充内容，吸引用户购买。 - 详细描述商品的优惠活动信息（如用户输入没有，不要询问，直接补充）。 - 在“购买方式”部分： - 仅包括引导用户点击链接购买的内容。 - 不要包括快递物流承诺。 - 【重要】每个部分多于5句话。 - 使用口语化的文字。 - 表达方式有亲和力和感染力。 - 对用户们使用“宝子们”、“宝宝们”、“家人们”等有亲和力的称呼。 - 多使用自问自答的语句。- 当你准备好了回复“请输入商品描述和优惠信息”我们就开始。#输出：- 包含在""里。- 写出每一部分内容的标题，包括【引出话题】、【产品讲解】、【购买方式】。- 示例：""【引出话题】【产品讲解】【购买方式】""- 输入商品信息:%s" % prompt # query = '请你扮演直播间直播卖货的主播角色，根据用户提出的问题，检索知识库中的信息，如果查找到相关的信息，使用口语化的文字，用非常简短的一句话回答用户的问题；如果查找不到相关的信息，直接回答没有，并引导用户关注本场直播商品。你的目的是激发用户的购买欲、提高直播商品销量。回答对用户们使用“宝子”、“宝宝”、“亲”等有亲和力的称呼。\n\n# 用户提出的问题：\n这里面有可乐吗\n\n# 知识库：\n商品名称：外送欢乐桶(券)兑换券\n直播链接序号：64号链接号链接\n商品现价：66元\n商品原价：142元\n折扣：5\n品牌名称：麦当劳麦乐送\n券包张数：1\n套餐详情：["麦辣鸡腿汉堡1个","板烧鸡腿堡1个","中可乐3杯","四拼小食桶A"]\n补充：四拼小食桶A(麦麦脆汁鸡(鸡腿)1块＋中薯条1份＋麦辣鸡翅2块＋麦乐鸡5块)＋麦辣鸡腿堡1个＋板烧鸡腿堡1个＋中可乐3杯 原材料：面包\n\n\n# 输出格式：\n答：\n\n# 要求\n对于用户问到了知识库中未提及的信息不要编造，直接不回答' inputs = [] ###### img_info = np.array([img.encode('utf-8')], dtype=np.string_) img_info = np.tile(img_info, (bs,1)) inputs.append(InferInput("img", img_info.shape, np_to_triton_dtype(img_info.dtype)),) inputs[0].set_data_from_numpy(img_info) ##### title_text = np.array([query.encode('utf-8')], dtype=np.string_) title_text = np.tile(title_text, (bs, 1)) data_batch = {} data_batch['query'] = title_text inputs.append( InferInput("query", data_batch['query'].shape, np_to_triton_dtype(data_batch['query'].dtype)) ) inputs[1].set_data_from_numpy(data_batch['query']) return inputs def make_a_try(inputs, outputs='response', model_name='ensemble_mllm', model_version=''): # qwen 1 outputs_list = [] ori_outputs_list = outputs.strip().split(",") for out_ele in ori_outputs_list: outputs_list.append(out_ele.strip()) outputs = [InferRequestedOutput(x) for x in outputs_list] response = my_client.predict(model_name=model_name, inputs=inputs, model_version=model_version, outputs=outputs) rsp_info = {} if outputs_list == []: for out_name_ele in response._result.outputs: outputs_list.append(out_name_ele.name) for output_name in outputs_list: res = response.as_numpy(output_name) response = np.expand_dims(res, axis=0) response = response[0].decode('utf-8') rsp_info[output_name] = response print("response:",rsp_info) return rsp_info['response'] # def greet(prompt): # """Greet someone.""" # # print(prompt) # print("prompt:",prompt) # inputs = prepare_params(prompt) # print(inputs) # result = make_a_try(inputs) # return result def greet(prompt,img): """Greet someone.""" # print(prompt) print("prompt:",prompt) inputs = prepare_params(prompt,img) print(inputs) result = make_a_try(inputs) return result def clear_input(): return "" if __name__ == "__main__": param_info = {} # param_info['appkey'] = "com.sankuai.automl.serving" param_info['appkey'] = "com.sankuai.automl.streamvlm" # param_info['remote_appkey'] = "com.sankuai.automl.chat3" param_info['remote_appkey'] = "com.sankuai.automl.streamvlm" param_info['model_name'] = "ensemble_mllm" param_info['model_version'] = "1" param_info['time_out'] = 60000 param_info['server_targets'] = [] param_info['outputs'] = 'response' appkey, remote_appkey, model_name, model_version, time_out, server_targets = param_info['appkey'], param_info['remote_appkey'], param_info['model_name'], param_info['model_version'], param_info['time_out'], param_info['server_targets'] my_client = TritonServerClient(appkey=appkey, remote_appkey=remote_appkey, time_out=time_out, server_targets=server_targets) # triton_client.async_stream_infer( # model_name=model_name, # inputs=inputs, # outputs=outputs, # request_id="{}".format(sequence_id), # sequence_id=sequence_id, # sequence_start=True, # sequence_end=True, # ) img_url = f"https://s3plus.sankuai.com/automl-pkgs/0000.jpeg" greet("nihao",img_url) # greet("nihao") print("描述这张图片")