Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| """ | |
| multimodal gpt-oss 120b — Gradio app for Hugging Face Spaces | |
| Функции: | |
| - Загрузка собственной картинки (type="filepath") | |
| - Галерея примеров (клик -> подставляет в загрузчик) | |
| - Автогенерация "More Detailed Caption" через gradio_client Florence-2 | |
| - Streaming ответов от openai/gpt-oss-120b (через NVIDIA integrate / OpenAI-compatible) | |
| - Кеширование подписи для одной и той же картинки | |
| """ | |
| import os | |
| import traceback | |
| from typing import Any, Dict, List, Tuple, Optional | |
| import gradio as gr | |
| from gradio_client import Client, handle_file | |
| from openai import OpenAI | |
| # (опционально) локальный .env при локальном запуске | |
| try: | |
| from dotenv import load_dotenv | |
| load_dotenv() | |
| except Exception: | |
| pass | |
| # --------------------- Конфигурация --------------------- | |
| NV_API_KEY = os.environ.get("NV_API_KEY") # ОБЯЗАТЕЛЬНО прописать в Secrets HF Spaces | |
| NV_BASE_URL = os.environ.get("NV_BASE_URL", "https://integrate.api.nvidia.com/v1") | |
| if not NV_API_KEY: | |
| raise RuntimeError( | |
| "NV_API_KEY не задан. В Hugging Face Space зайди в Settings → Secrets и добавь NV_API_KEY." | |
| ) | |
| # Florence-2 Gradio wrapper (публичный) | |
| FLORENCE_WRAPPER = "gokaygokay/Florence-2" | |
| # --------------------- Клиенты --------------------- | |
| florence = Client(FLORENCE_WRAPPER) | |
| llm = OpenAI(base_url=NV_BASE_URL, api_key=NV_API_KEY) | |
| # --------------------- Хелперы --------------------- | |
| def _normalize_florence_result(res: Any) -> str: | |
| """ | |
| Нормализует результат predict от Florence-2: возвращает текстовую подпись. | |
| Подстраховываемся на разные форматы (строка, dict, list и т.д.). | |
| """ | |
| try: | |
| if res is None: | |
| return "" | |
| if isinstance(res, str): | |
| return res | |
| # dict-like | |
| if isinstance(res, dict): | |
| # часто бывает ключ 'caption' или 'text' или 'generated_text' | |
| for k in ("caption", "text", "generated_text", "output", "result"): | |
| if k in res and isinstance(res[k], str): | |
| return res[k] | |
| # если есть nested fields, попробуем взять первое строковое значение | |
| for v in res.values(): | |
| if isinstance(v, str): | |
| return v | |
| # fallback: str() | |
| return str(res) | |
| # list/tuple: join string elements | |
| if isinstance(res, (list, tuple)): | |
| pieces = [str(x) for x in res] | |
| return "\n".join(pieces) | |
| # other types: fallback to str | |
| return str(res) | |
| except Exception: | |
| return f"[Ошибка нормализации подписи: {traceback.format_exc()}]" | |
| def get_caption_for_image(image_path_or_url: str, safety_note: bool = False) -> str: | |
| """ | |
| Запрос к Florence-2: task_prompt="More Detailed Caption". | |
| Принимает локальный путь или URL. | |
| """ | |
| try: | |
| if not image_path_or_url: | |
| return "" | |
| # handle_file поддерживает URL и локальные пути | |
| res = florence.predict( | |
| image=handle_file(image_path_or_url), | |
| task_prompt="More Detailed Caption", | |
| text_input=None, | |
| model_id="microsoft/Florence-2-large", | |
| api_name="/process_image" | |
| ) | |
| caption = _normalize_florence_result(res) | |
| return caption | |
| except Exception as e: | |
| # логируем в stdout (HF Spaces покажет лог) | |
| print("Ошибка Florence-2 predict:", e) | |
| traceback.print_exc() | |
| return f"[Ошибка при генерации подписи: {e}]" | |
| def _extract_text_from_stream_chunk(chunk: Any) -> str: | |
| """ | |
| Универсально извлекает текстовые фрагменты из чанка стриминга LLM. | |
| Работает с разными формами chunk (объект SDK, dict, ...) | |
| """ | |
| try: | |
| # объектный стиль: chunk.choices[0].delta.content | |
| if hasattr(chunk, "choices"): | |
| choices = getattr(chunk, "choices") | |
| if choices: | |
| c0 = choices[0] | |
| delta = getattr(c0, "delta", None) | |
| if delta is not None: | |
| # reasoning_content или content | |
| txt = getattr(delta, "reasoning_content", None) or getattr(delta, "content", None) | |
| if txt: | |
| return str(txt) | |
| # some SDK might put content in c0.get("text") etc. | |
| text_attr = getattr(c0, "text", None) | |
| if text_attr: | |
| return str(text_attr) | |
| # dict-style | |
| if isinstance(chunk, dict): | |
| choices = chunk.get("choices") or [] | |
| if choices: | |
| delta = choices[0].get("delta") or {} | |
| # try common keys | |
| return str(delta.get("content") or delta.get("reasoning_content") or choices[0].get("text") or "") | |
| except Exception: | |
| pass | |
| return "" | |
| # --------------------- UI-логика --------------------- | |
| # Кеш подписи (чтобы не вызывать Florence снова для той же картинки) | |
| # Храним словарь: {"image_path": "...", "caption": "..."} | |
| # Будем использовать gr.State для хранения этого словаря в сессии | |
| def generate_and_cache_caption(image, cache: Optional[Dict[str, str]]): | |
| """ | |
| Вызывается при изменении image_input или при клике по галерее. | |
| Возвращает (caption_text, new_cache_dict). | |
| """ | |
| try: | |
| if not image: | |
| return "", {"image_path": None, "caption": None} | |
| # Готовим path/URL | |
| img_path = image if isinstance(image, str) else getattr(image, "name", None) or image | |
| # Проверка кеша | |
| if cache and cache.get("image_path") == img_path and cache.get("caption"): | |
| return cache.get("caption"), cache | |
| # Иначе генерируем подпись | |
| caption = get_caption_for_image(img_path) | |
| new_cache = {"image_path": img_path, "caption": caption} | |
| return caption, new_cache | |
| except Exception as e: | |
| print("generate_and_cache_caption exception:", e) | |
| traceback.print_exc() | |
| return f"[Ошибка генерации подписи: {e}]", {"image_path": None, "caption": None} | |
| def chat_stream(image, user_message, history, cache: Dict[str, str]): | |
| """ | |
| Основной generator для кнопки Отправить / submit: | |
| - Автоматически использует кеш подписи (если есть), иначе генерирует новую | |
| - Возвращает по мере стриминга (history, caption) — соответствие outputs=[chatbot, raw_caption] | |
| """ | |
| history = history or [] | |
| # Проверки входа | |
| if not user_message: | |
| # ничего не делаем, просто возвращаем текущее состояние | |
| yield history, (cache.get("caption") if cache else "") | |
| return | |
| if not image: | |
| # если нет картинки — говорим пользователю | |
| history.append([user_message, "Пожалуйста, загрузите изображение или выберите из галереи."]) | |
| yield history, (cache.get("caption") if cache else "") | |
| return | |
| # получить путь и подпись (используем кеш, если совпадает) | |
| img_path = image if isinstance(image, str) else getattr(image, "name", None) or image | |
| if cache and cache.get("image_path") == img_path and cache.get("caption"): | |
| caption = cache.get("caption") | |
| else: | |
| caption = get_caption_for_image(img_path) | |
| # обновляем кеш локально (не gr.State, а для текущего запроса) | |
| cache = {"image_path": img_path, "caption": caption} | |
| # система-промпт — даём контекст и просим указывать степень уверенности | |
| system_prompt = ( | |
| "You are 'multimodal gpt-oss 120b', a helpful multimodal assistant. " | |
| "Use the provided 'More Detailed Caption' as authoritative visual context. " | |
| "If something is not visible or certain, say so explicitly.\n\n" | |
| "Image Caption START >>>\n" | |
| f"{caption}\n" | |
| "<<< Image Caption END.\n" | |
| "Answer the user's question based on the caption and general knowledge. " | |
| "Be concise unless asked for details." | |
| ) | |
| # добавляем пользовательский запрос в историю (пустой ответ пока) | |
| history.append([user_message, ""]) # assistant текст будет заполняться по мере стрима | |
| # первый yield чтобы UI сразу отобразил user's message и подпись | |
| yield history, caption | |
| assistant_accum = "" | |
| try: | |
| # Запускаем стриминг вызов | |
| stream = llm.chat.completions.create( | |
| model="openai/gpt-oss-120b", | |
| messages=[ | |
| {"role": "system", "content": system_prompt}, | |
| {"role": "user", "content": user_message} | |
| ], | |
| temperature=0.8, | |
| top_p=1.0, | |
| max_tokens=1024, | |
| stream=True, | |
| ) | |
| for chunk in stream: | |
| piece = _extract_text_from_stream_chunk(chunk) | |
| if not piece: | |
| continue | |
| assistant_accum += piece | |
| # обновляем последний элемент истории (assistant part) | |
| history[-1][1] = assistant_accum | |
| yield history, caption | |
| except Exception as e: | |
| # Ошибка стриминга: попробуем получить финальный ответ без стрима, либо показать ошибку | |
| print("Streaming error:", e) | |
| traceback.print_exc() | |
| # Пытаемся сделать не-стриминг вызов (fallback) | |
| try: | |
| resp = llm.chat.completions.create( | |
| model="openai/gpt-oss-120b", | |
| messages=[ | |
| {"role": "system", "content": system_prompt}, | |
| {"role": "user", "content": user_message} | |
| ], | |
| temperature=0.8, | |
| top_p=1.0, | |
| max_tokens=1024, | |
| stream=False, | |
| ) | |
| # нормализуем возможный формат ответа | |
| final_text = "" | |
| # SDK может вернуть object-like resp.choices[0].message.content | |
| if hasattr(resp, "choices"): | |
| try: | |
| final_text = getattr(resp.choices[0].message, "content", "") or getattr(resp.choices[0], "text", "") or "" | |
| except Exception: | |
| final_text = str(resp) | |
| elif isinstance(resp, dict): | |
| choices = resp.get("choices", []) | |
| if choices: | |
| m = choices[0].get("message") or choices[0] | |
| final_text = m.get("content") or m.get("text") or str(m) | |
| else: | |
| final_text = str(resp) | |
| else: | |
| final_text = str(resp) | |
| history[-1][1] = final_text | |
| yield history, caption | |
| except Exception as e2: | |
| history[-1][1] = f"[Ошибка LLM: {e2}]" | |
| yield history, caption | |
| # финальный yield (гарантируем окончательное состояние) | |
| yield history, caption | |
| # --------------------- Примеры для галереи (список строк) --------------------- | |
| EXAMPLE_IMAGES = [ | |
| "https://raw.githubusercontent.com/gradio-app/gradio/main/test/test_files/bus.png", | |
| "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/cats.png", | |
| "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/cheetah.jpg", | |
| "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/flowers.png", | |
| ] | |
| # --------------------- UI --------------------- | |
| css = """ | |
| .gradio-container { max-width: 1100px; margin: auto; } | |
| #title { text-align: center; } | |
| """ | |
| with gr.Blocks(css=css, analytics_enabled=False) as demo: | |
| gr.Markdown("<h2 id='title'>🖼️ multimodal gpt-oss 120b — визуальный чат</h2>") | |
| with gr.Row(): | |
| with gr.Column(scale=4): | |
| image_input = gr.Image(label="Загрузите картинку (файл / drag-n-drop / камера)", type="filepath") | |
| raw_caption = gr.Textbox(label="More Detailed Caption (Florence-2)", interactive=False, lines=6) | |
| user_input = gr.Textbox(label="Вопрос по изображению", placeholder="Например: Что происходит на фото?") | |
| send_btn = gr.Button("Отправить") | |
| clear_btn = gr.Button("Очистить чат") | |
| gr.Markdown("**Галерея примеров (клик — подставить в загрузчик и получить подпись)**") | |
| gallery = gr.Gallery(value=EXAMPLE_IMAGES, label="Примеры", columns=4, show_label=False).style(grid=[4]) | |
| with gr.Column(scale=6): | |
| chatbot = gr.Chatbot(label="Чат с моделью", height=640) | |
| # gr.State для кеша подписи | |
| caption_cache = gr.State(value={"image_path": None, "caption": None}) | |
| # обработчик клика по галерее: сразу подставляет картинку, генерирует подпись и обновляет кеш | |
| def on_gallery_select(elem, cach | |