main.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 make_a_try(inputs, outputs='response', model_name='qwen', model_version='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
   
    return rsp_info['response']



def greet(prompt):
    """Greet someone."""
    # print(prompt)
    inputs = prepare_params(prompt)
    result = make_a_try(inputs)

    return result


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['model_name'] = "qwen"
    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)

   # 以上部分，用户可以修改代码
    demo = gr.Interface(
        fn=greet,
        inputs=["textbox"],
        outputs=["textbox"],
    )

    demo.launch(server_name="0.0.0.0", server_port=8088, debug=True, share=True)