Spaces:
Build error
Build error
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 | |
import tritonclient.grpc as grpcclient | |
from tritonclient.utils import InferenceServerException | |
# 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("描述这张图片") |