try_hd.py

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("描述这张图片")