Spaces:
Running
Running
File size: 2,662 Bytes
d3cd5c1 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 |
import argparse
import time
from threading import Thread
import gradio as gr
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer
from moondream.hf import LATEST_REVISION, detect_device
parser = argparse.ArgumentParser()
parser.add_argument("--cpu", action="store_true")
args = parser.parse_args()
if args.cpu:
device = torch.device("cpu")
dtype = torch.float32
else:
device, dtype = detect_device()
if device != torch.device("cpu"):
print("Using device:", device)
print("If you run into issues, pass the `--cpu` flag to this script.")
print()
model_id = "vikhyatk/moondream2"
tokenizer = AutoTokenizer.from_pretrained(model_id, revision=LATEST_REVISION)
moondream = AutoModelForCausalLM.from_pretrained(
model_id, trust_remote_code=True, revision=LATEST_REVISION
).to(device=device, dtype=dtype)
moondream.eval()
def answer_question(img, prompt):
image_embeds = moondream.encode_image(img)
streamer = TextIteratorStreamer(tokenizer, skip_special_tokens=True)
thread = Thread(
target=moondream.answer_question,
kwargs={
"image_embeds": image_embeds,
"question": prompt,
"tokenizer": tokenizer,
"streamer": streamer,
},
)
thread.start()
buffer = ""
for new_text in streamer:
buffer += new_text
yield buffer
with gr.Blocks() as demo:
gr.Markdown("# See For Me")
gr.HTML(
"""
<style type="text/css">
.md_output p {
padding-top: 1rem;
font-size: 1.2rem !important;
}
</style>
"""
)
with gr.Row():
prompt = gr.Textbox(
label="Prompt",
value="What's going on? Respond with a single sentence.",
interactive=True,
)
with gr.Row():
img = gr.Image(type="pil", label="Upload an Image", streaming=True)
output = gr.Markdown(elem_classes=["md_output"])
latest_img = None
latest_prompt = prompt.value
@img.change(inputs=[img])
def img_change(img):
global latest_img
latest_img = img
@prompt.change(inputs=[prompt])
def prompt_change(prompt):
global latest_prompt
latest_prompt = prompt
@demo.load(outputs=[output])
def live_video():
while True:
if latest_img is None:
time.sleep(0.1)
else:
for text in answer_question(latest_img, latest_prompt):
if len(text) > 0:
yield text
demo.queue().launch(debug=True)
|