import osimport sysimport randomimport uuidimport jsonimport timefrom threading import Threadfrom typing import Iterablefrom huggingface_hub import snapshot_download import gradio as grimport torchimport numpy as npfrom PIL import Imageimport cv2 from transformers import ( AutoModelForCausalLM, AutoProcessor, TextIteratorStreamer,) from transformers.image_utils import load_image css = """#main-title h1 { font-size: 2.3em !important;}#output-title h2 { font-size: 2.1em !important;}""" MAX_MAX_NEW_TOKENS = 4096DEFAULT_MAX_NEW_TOKENS = 2048MAX_INPUT_TOKEN_LENGTH = int(os.getenv("MAX_INPUT_TOKEN_LENGTH", "4096")) device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu") print("--- System Information ---")print("CUDA_VISIBLE_DEVICES=", os.environ.get("CUDA_VISIBLE_DEVICES"))print("torch.__version__ =", torch.__version__)print("torch.version.cuda =", torch.version.cuda)print("CUDA available:", torch.cuda.is_available())print("CUDA device count:", torch.cuda.device_count())if torch.cuda.is_available(): print("Current device:", torch.cuda.current_device()) print("Device name:", torch.cuda.get_device_name(torch.cuda.current_device()))print("Using device:", device)print("--------------------------") print("Loading Dots.OCR model...")MODEL_PATH_D = "prithivMLmods/Dots.OCR-Latest-BF16"processor = AutoProcessor.from_pretrained(MODEL_PATH_D, trust_remote_code=True)model = AutoModelForCausalLM.from_pretrained( MODEL_PATH_D, attn_implementation="flash_attention_2", torch_dtype=torch.bfloat16, device_map="auto", trust_remote_code=True).eval()print("Dots.OCR model loaded successfully.") def generate_image(text: str, image: Image.Image, max_new_tokens: int, temperature: float, top_p: float, top_k: int, repetition_penalty: float): """ Generates responses using the Dots.OCR model for image input. Yields raw text and Markdown-formatted text. """ if image is None: yield "Please upload an image.", "Please upload an image." return messages = [{ "role": "user", "content": [ {"type": "image"}, {"type": "text", "text": text}, ] }] prompt_full = processor.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) inputs = processor( text=[prompt_full], images=[image], return_tensors="pt", padding=True).to(device) streamer = TextIteratorStreamer(processor, skip_prompt=True, skip_special_tokens=True) generation_kwargs = { **inputs, "streamer": streamer, "max_new_tokens": max_new_tokens, "do_sample": True, "temperature": temperature, "top_p": top_p, "top_k": top_k, "repetition_penalty": repetition_penalty, } thread = Thread(target=model.generate, kwargs=generation_kwargs) thread.start() buffer = "" for new_text in streamer: buffer += new_text # Clean up potential end-of-sequence tokens from the buffer buffer = buffer.replace("<|im_end|>", "") time.sleep(0.01) yield buffer, buffer with gr.Blocks(css=css) as demo: gr.Markdown("# **Dots.OCR**", elem_id="main-title") with gr.Row(): with gr.Column(scale=2): image_query = gr.Textbox(label="Query Input", placeholder="Enter your query here...") image_upload = gr.Image(type="pil", label="Upload Image", height=290) image_submit = gr.Button("Submit", variant="primary") with gr.Accordion("Advanced options", open=False): max_new_tokens = gr.Slider(label="Max new tokens", minimum=1, maximum=MAX_MAX_NEW_TOKENS, step=1, value=DEFAULT_MAX_NEW_TOKENS) temperature = gr.Slider(label="Temperature", minimum=0.1, maximum=4.0, step=0.1, value=0.7) top_p = gr.Slider(label="Top-p (nucleus sampling)", minimum=0.05, maximum=1.0, step=0.05, value=0.9) top_k = gr.Slider(label="Top-k", minimum=1, maximum=1000, step=1, value=50) repetition_penalty = gr.Slider(label="Repetition penalty", minimum=1.0, maximum=2.0, step=0.05, value=1.1) with gr.Column(scale=3): gr.Markdown("## Output", elem_id="output-title") output = gr.Textbox(label="Raw Output Stream", interactive=False, lines=15, show_copy_button=True) with gr.Accordion("(Result.md)", open=False): markdown_output = gr.Markdown(label="(Result.Md)") image_submit.click( fn=generate_image, inputs=[image_query, image_upload, max_new_tokens, temperature, top_p, top_k, repetition_penalty], outputs=[output, markdown_output] ) if __name__ == "__main__": demo.queue(max_size=50).launch(ssr_mode=False, show_error=True)