# Copyright (c) Alibaba, Inc. and its affiliates. import os import secrets import tempfile from http import HTTPStatus from pathlib import Path import gradio as gr import modelscope_studio.components.antd as antd import modelscope_studio.components.base as ms from PIL import Image from urllib3.exceptions import HTTPError os.environ['DASHSCOPE_HTTP_BASE_URL'] = 'https://dashscope.aliyuncs.com/api/v1' # os.environ['DASHSCOPE_WEBSOCKET_BASE_URL'] = 'https://poc-dashscope.aliyuncs.com/api-ws/v1/inference' import dashscope from dashscope import MultiModalConversation API_KEY = os.environ.get('API_KEY') dashscope.api_key = API_KEY is_modelscope_studio = os.getenv('MODELSCOPE_ENVIRONMENT') == 'studio' def get_text(text: str, cn_text: str): if is_modelscope_studio: return cn_text return text def resolve_image(filename): return os.path.join(os.path.dirname(__file__), filename) DEMO_LIST = [ { "description": "Evaluate the integral of the functions graphed using the formula for circles: ", "image": resolve_image("./examples/1.webp") }, { "description": "请解答这道题", "image": resolve_image("./examples/5.png") }, { "description": "图片中的滤液E是什么化学物质?", "image": resolve_image("./examples/3.png") }, { "description": "How many pelicans are there in the picture", "image": resolve_image("./examples/6.png") }, ] def process_image(image, shouldConvert=False): # 获取上传文件的目录 uploaded_file_dir = os.environ.get("GRADIO_TEMP_DIR") or str( Path(tempfile.gettempdir()) / "gradio") os.makedirs(uploaded_file_dir, exist_ok=True) # 创建临时文件路径 name = f"tmp{secrets.token_hex(20)}.jpg" filename = os.path.join(uploaded_file_dir, name) # 保存上传的图片 if shouldConvert: new_img = Image.new('RGB', size=(image.width, image.height), color=(255, 255, 255)) new_img.paste(image, (0, 0), mask=image) image = new_img image.save(filename) return filename def on_clear(): return { input: gr.update(value=None), **{ item: gr.update(value=None) for item in input_image }, } css = """ .output-markdown { overflow: unset !important; } """ with gr.Blocks(css=css) as demo: with ms.Application() as app: with antd.ConfigProvider( locale="zh_CN" if is_modelscope_studio else None, theme=dict(token=dict(colorPrimary="#a855f7"))): with antd.Card(elem_style=dict(marginBottom=12), styles=dict(body=dict(padding=4))): with antd.Flex(elem_style=dict(width="100%"), justify="center", align="center", gap=14): with ms.Div(elem_style=dict(flexShrink=0)): antd.Image(resolve_image("./cutelogo.jpg"), preview=False, height=60, width=60) with ms.Div(): antd.Typography.Title("QVQ-72B-Preview", elem_style=dict(margin=0, fontSize=24), level=1) with ms.AutoLoading(): with antd.Row(gutter=[8, 8], align="stretch"): with antd.Col(xs=24, md=8): with antd.Space(direction="vertical", elem_style=dict(width="100%")): with antd.Space(direction="vertical", elem_style=dict(width="100%"), elem_id="input-container"): with ms.Fragment(): input_image = gr.Image(type="pil", label="Upload", sources=["upload"]), input = antd.Input.Textarea( placeholder=get_text( "Ask a question", "输入一个问题"), auto_size=dict(maxRows=6, minRows=2), allow_clear=True) with antd.Flex(align="center", justify="space-between"): antd.Typography.Text(get_text( "Warning: This model only supports single-turn dialogue.", "注：当前模型只支持单轮对话，如需中文回答，提示词加“用中文回答”"), type="warning") tour_btn = antd.Button(get_text( "Tour", "使用指引"), variant="filled", color="default") with antd.Row(gutter=8): with antd.Col(span=12): clear_btn = antd.Button(get_text( "Clear", "清除"), block=True) with antd.Col(span=12): submit_btn = antd.Button( get_text("Submit", "提交"), type="primary", block=True, elem_id="submit-btn") antd.Divider(get_text("Examples", "示例")) with antd.Flex(gap="small", wrap=True): for item in DEMO_LIST: def bind_on_example(_item): def on_example(): return gr.update( value=_item['description'] ), gr.update(value=_item['image']) return on_example with antd.Card( hoverable=True, elem_style=dict( width="100%")) as example: if "description" in item: antd.Typography.Text( item["description"]) if "image" in item: antd.Image(item["image"], preview=False) example.click( fn=bind_on_example(item), outputs=[input, input_image[0]]) with antd.Col(xs=24, md=16): with antd.Card(title=get_text("Answer", "答案"), elem_style=dict(height="100%"), elem_id="output-container"): with ms.Slot("extra"): cancel_btn = antd.Button(get_text( "Stop", "停止"), elem_id="cancel-btn", block=True, disabled=True) with ms.Div(elem_style=dict( maxHeight=1600, display="flex", flexDirection="column-reverse", overflow="auto")): output = gr.Markdown( show_copy_button=True, elem_classes="output-markdown", latex_delimiters=[{ "left": '$$', "right": '$$', "display": True }, { "left": '$', "right": '$', "display": False, }, { "left": '\\(', "right": '\\)', "display": False, }, { "left": '\\[', "right": '\\]', "display": True }]) with antd.Tour(open=False) as tour: antd.Tour.Step( title=get_text("Step 1", "步骤 1"), description=get_text("Upload image and enter text", "传入图片和文本"), get_target= "() => document.querySelector('#input-container')") antd.Tour.Step( title=get_text("Step 2", "步骤 2"), description=get_text("Click the submit button", "点击提交按钮"), get_target="() => document.querySelector('#submit-btn')" ) antd.Tour.Step( title=get_text("Step 3", "步骤 3"), description=get_text("Wait for the result", "等待结果返回"), get_target= "() => document.querySelector('#output-container')") antd.Tour.Step( title=get_text("Tips", "提示"), description=get_text("Click here to end output early", "点击这里提前结束输出"), get_target="() => document.querySelector('#cancel-btn')" ) tour_btn.click(fn=lambda: gr.update(open=True), outputs=[tour]) gr.on([tour.finish, tour.close], fn=lambda: gr.update(open=False), outputs=[tour]) def generate(image, query): content = [] if not image and not query: raise gr.Error( get_text("Error: Input is empty", "错误：输入内容为空")) if image: imageFile = process_image(image) content.append({'image': f'file://{imageFile}'}) if query: content.append({'text': query}) print("image", image) print("query", query) messages = [ { 'role': 'user', 'content': content }, ] responses = MultiModalConversation.call( model='qvq-72b-preview', messages=messages, stream=True, ) yield {cancel_btn: gr.update(disabled=False)} for response in responses: if not response.status_code == HTTPStatus.OK: raise HTTPError( f'response.code: {response.code}\nresponse.message: {response.message}' ) response = response.output.choices[0].message.content if len(response) > 0 and response[0]['text']: print(response[0]['text']) yield {output: response[0]['text']} yield {cancel_btn: gr.update(disabled=True)} output_process = submit_btn.click(fn=generate, inputs=[*input_image, input], outputs=[output, cancel_btn]) clear_btn.click(fn=on_clear, outputs=[*input_image, input]) cancel_btn.click(fn=None, inputs=None, outputs=None, cancels=[output_process]) cancel_btn.click(fn=lambda: gr.update(disabled=True), inputs=None, outputs=[cancel_btn]) demo.queue(default_concurrency_limit=50).launch(ssr_mode=False)