app.py

#!/usr/bin/env python

import os
import re
import tempfile
import gc  # Added garbage collector
from collections.abc import Iterator
from threading import Thread
import json
import requests
import cv2
import base64
import logging
import time
from urllib.parse import quote  # Added for URL encoding
import importlib  # For dynamic import

import gradio as gr
import spaces
import torch
from loguru import logger
from PIL import Image
from transformers import AutoProcessor, Gemma3ForConditionalGeneration, TextIteratorStreamer

# CSV/TXT/PDF analysis
import pandas as pd
import PyPDF2

# =============================================================================
# (New) Image API related functions
# =============================================================================
from gradio_client import Client

API_URL = "http://211.233.58.201:7896"

logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

def test_api_connection() -> str:
    """Test API server connection"""
    try:
        client = Client(API_URL)
        return "API connection successful: Operating normally"
    except Exception as e:
        logging.error(f"API connection test failed: {e}")
        return f"API connection failed: {e}"

def generate_image(prompt: str, width: float, height: float, guidance: float, inference_steps: float, seed: float):
    """Image generation function (flexible return types)"""
    if not prompt:
        return None, "Error: A prompt is required."
    try:
        logging.info(f"Calling image generation API with prompt: {prompt}")
        
        client = Client(API_URL)
        result = client.predict(
            prompt=prompt,
            width=int(width),
            height=int(height),
            guidance=float(guidance),
            inference_steps=int(inference_steps),
            seed=int(seed),
            do_img2img=False,
            init_image=None,
            image2image_strength=0.8,
            resize_img=True,
            api_name="/generate_image"
        )
        
        logging.info(f"Image generation result: {type(result)}, length: {len(result) if isinstance(result, (list, tuple)) else 'unknown'}")
        
        # Handle cases where the result is a tuple or list
        if isinstance(result, (list, tuple)) and len(result) > 0:
            image_data = result[0]  # The first element is the image data
            seed_info = result[1] if len(result) > 1 else "Unknown seed"
            return image_data, seed_info
        else:
            # When a single value is returned
            return result, "Unknown seed"
            
    except Exception as e:
        logging.error(f"Image generation failed: {str(e)}")
        return None, f"Error: {str(e)}"

def fix_base64_padding(data):
    """Fix the padding of a Base64 string."""
    if isinstance(data, bytes):
        data = data.decode('utf-8')
    
    if "base64," in data:
        data = data.split("base64,", 1)[1]
    
    missing_padding = len(data) % 4
    if missing_padding:
        data += '=' * (4 - missing_padding)
    
    return data

def clear_cuda_cache():
    """Explicitly clear the CUDA cache."""
    if torch.cuda.is_available():
        torch.cuda.empty_cache()
        gc.collect()

SERPHOUSE_API_KEY = os.getenv("SERPHOUSE_API_KEY", "")

def extract_keywords(text: str, top_k: int = 5) -> str:
    """Simple keyword extraction: only keep English, Korean, numbers, and spaces."""
    text = re.sub(r"[^a-zA-Z0-9가-힣\s]", "", text)
    tokens = text.split()
    return " ".join(tokens[:top_k])

def do_web_search(query: str) -> str:
    """Call the SerpHouse LIVE API to return Markdown formatted search results"""
    try:
        url = "https://api.serphouse.com/serp/live"
        params = {
            "q": query,
            "domain": "google.com",
            "serp_type": "web",
            "device": "desktop",
            "lang": "en",
            "num": "20"
        }
        headers = {"Authorization": f"Bearer {SERPHOUSE_API_KEY}"}
        logger.info(f"Calling SerpHouse API with query: {query}")
        response = requests.get(url, headers=headers, params=params, timeout=60)
        response.raise_for_status()
        data = response.json()
        results = data.get("results", {})
        organic = None
        if isinstance(results, dict) and "organic" in results:
            organic = results["organic"]
        elif isinstance(results, dict) and "results" in results:
            if isinstance(results["results"], dict) and "organic" in results["results"]:
                organic = results["results"]["organic"]
        elif "organic" in data:
            organic = data["organic"]
        if not organic:
            logger.warning("Organic results not found in response.")
            return "No web search results available or the API response structure is unexpected."
        max_results = min(20, len(organic))
        limited_organic = organic[:max_results]
        summary_lines = []
        for idx, item in enumerate(limited_organic, start=1):
            title = item.get("title", "No Title")
            link = item.get("link", "#")
            snippet = item.get("snippet", "No Description")
            displayed_link = item.get("displayed_link", link)
            summary_lines.append(
                f"### Result {idx}: {title}\n\n"
                f"{snippet}\n\n"
                f"**Source**: [{displayed_link}]({link})\n\n"
                f"---\n"
            )
        instructions = """
# Web Search Results
Below are the search results. Use this information to answer the query:
1. Refer to each result's title, description, and source link.
2. In your answer, explicitly cite the source of any used information (e.g., "[Source Title](link)").
3. Include the actual source links in your response.
4. Synthesize information from multiple sources.
5. At the end include a "References:" section listing the main source links.
"""
        return instructions + "\n".join(summary_lines)
    except Exception as e:
        logger.error(f"Web search failed: {e}")
        return f"Web search failed: {str(e)}"

MAX_CONTENT_CHARS = 2000
MAX_INPUT_LENGTH = 2096
model_id = os.getenv("MODEL_ID", "VIDraft/Gemma-3-R1984-4B")
processor = AutoProcessor.from_pretrained(model_id, padding_side="left")
model = Gemma3ForConditionalGeneration.from_pretrained(
    model_id,
    device_map="auto",
    torch_dtype=torch.bfloat16,
    attn_implementation="eager"
)
MAX_NUM_IMAGES = int(os.getenv("MAX_NUM_IMAGES", "5"))

def analyze_csv_file(path: str) -> str:
    try:
        df = pd.read_csv(path)
        if df.shape[0] > 50 or df.shape[1] > 10:
            df = df.iloc[:50, :10]
        df_str = df.to_string()
        if len(df_str) > MAX_CONTENT_CHARS:
            df_str = df_str[:MAX_CONTENT_CHARS] + "\n...(truncated)..."
        return f"**[CSV File: {os.path.basename(path)}]**\n\n{df_str}"
    except Exception as e:
        return f"CSV file read failed ({os.path.basename(path)}): {str(e)}"

def analyze_txt_file(path: str) -> str:
    try:
        with open(path, "r", encoding="utf-8") as f:
            text = f.read()
        if len(text) > MAX_CONTENT_CHARS:
            text = text[:MAX_CONTENT_CHARS] + "\n...(truncated)..."
        return f"**[TXT File: {os.path.basename(path)}]**\n\n{text}"
    except Exception as e:
        return f"TXT file read failed ({os.path.basename(path)}): {str(e)}"

def pdf_to_markdown(pdf_path: str) -> str:
    text_chunks = []
    try:
        with open(pdf_path, "rb") as f:
            reader = PyPDF2.PdfReader(f)
            max_pages = min(5, len(reader.pages))
            for page_num in range(max_pages):
                page_text = reader.pages[page_num].extract_text() or ""
                page_text = page_text.strip()
                if page_text:
                    if len(page_text) > MAX_CONTENT_CHARS // max_pages:
                        page_text = page_text[:MAX_CONTENT_CHARS // max_pages] + "...(truncated)"
                    text_chunks.append(f"## Page {page_num+1}\n\n{page_text}\n")
            if len(reader.pages) > max_pages:
                text_chunks.append(f"\n...(Displaying only {max_pages} out of {len(reader.pages)} pages)...")
    except Exception as e:
        return f"PDF file read failed ({os.path.basename(pdf_path)}): {str(e)}"
    full_text = "\n".join(text_chunks)
    if len(full_text) > MAX_CONTENT_CHARS:
        full_text = full_text[:MAX_CONTENT_CHARS] + "\n...(truncated)..."
    return f"**[PDF File: {os.path.basename(pdf_path)}]**\n\n{full_text}"

def count_files_in_new_message(paths: list[str]) -> tuple[int, int]:
    image_count = 0
    video_count = 0
    for path in paths:
        if path.endswith(".mp4"):
            video_count += 1
        elif re.search(r"\.(png|jpg|jpeg|gif|webp)$", path, re.IGNORECASE):
            image_count += 1
    return image_count, video_count

def count_files_in_history(history: list[dict]) -> tuple[int, int]:
    image_count = 0
    video_count = 0
    for item in history:
        if item["role"] != "user" or isinstance(item["content"], str):
            continue
        if isinstance(item["content"], list) and len(item["content"]) > 0:
            file_path = item["content"][0]
            if isinstance(file_path, str):
                if file_path.endswith(".mp4"):
                    video_count += 1
                elif re.search(r"\.(png|jpg|jpeg|gif|webp)$", file_path, re.IGNORECASE):
                    image_count += 1
    return image_count, video_count

def validate_media_constraints(message: dict, history: list[dict]) -> bool:
    media_files = [f for f in message["files"] if re.search(r"\.(png|jpg|jpeg|gif|webp)$", f, re.IGNORECASE) or f.endswith(".mp4")]
    new_image_count, new_video_count = count_files_in_new_message(media_files)
    history_image_count, history_video_count = count_files_in_history(history)
    image_count = history_image_count + new_image_count
    video_count = history_video_count + new_video_count
    if video_count > 1:
        gr.Warning("Only one video file is supported.")
        return False
    if video_count == 1:
        if image_count > 0:
            gr.Warning("Mixing images and a video is not allowed.")
            return False
        if "<image>" in message["text"]:
            gr.Warning("The <image> tag cannot be used together with a video file.")
            return False
    if video_count == 0 and image_count > MAX_NUM_IMAGES:
        gr.Warning(f"You can upload a maximum of {MAX_NUM_IMAGES} images.")
        return False
    if "<image>" in message["text"]:
        image_files = [f for f in message["files"] if re.search(r"\.(png|jpg|jpeg|gif|webp)$", f, re.IGNORECASE)]
        image_tag_count = message["text"].count("<image>")
        if image_tag_count != len(image_files):
            gr.Warning("The number of <image> tags does not match the number of image files provided.")
            return False
    return True

def downsample_video(video_path: str) -> list[tuple[Image.Image, float]]:
    vidcap = cv2.VideoCapture(video_path)
    fps = vidcap.get(cv2.CAP_PROP_FPS)
    total_frames = int(vidcap.get(cv2.CAP_PROP_FRAME_COUNT))
    frame_interval = max(int(fps), int(total_frames / 10))
    frames = []
    for i in range(0, total_frames, frame_interval):
        vidcap.set(cv2.CAP_PROP_POS_FRAMES, i)
        success, image = vidcap.read()
        if success:
            image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
            image = cv2.resize(image, (0, 0), fx=0.5, fy=0.5)
            pil_image = Image.fromarray(image)
            timestamp = round(i / fps, 2)
            frames.append((pil_image, timestamp))
            if len(frames) >= 5:
                break
    vidcap.release()
    return frames

def process_video(video_path: str) -> tuple[list[dict], list[str]]:
    content = []
    temp_files = []
    frames = downsample_video(video_path)
    for pil_image, timestamp in frames:
        with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as temp_file:
            pil_image.save(temp_file.name)
            temp_files.append(temp_file.name)
            content.append({"type": "text", "text": f"Frame {timestamp}:"})
            content.append({"type": "image", "url": temp_file.name})
    return content, temp_files

def process_interleaved_images(message: dict) -> list[dict]:
    parts = re.split(r"(<image>)", message["text"])
    content = []
    image_files = [f for f in message["files"] if re.search(r"\.(png|jpg|jpeg|gif|webp)$", f, re.IGNORECASE)]
    image_index = 0
    for part in parts:
        if part == "<image>" and image_index < len(image_files):
            content.append({"type": "image", "url": image_files[image_index]})
            image_index += 1
        elif part.strip():
            content.append({"type": "text", "text": part.strip()})
        else:
            if isinstance(part, str) and part != "<image>":
                content.append({"type": "text", "text": part})
    return content

def is_image_file(file_path: str) -> bool:
    return bool(re.search(r"\.(png|jpg|jpeg|gif|webp)$", file_path, re.IGNORECASE))

def is_video_file(file_path: str) -> bool:
    return file_path.endswith(".mp4")

def is_document_file(file_path: str) -> bool:
    return file_path.lower().endswith(".pdf") or file_path.lower().endswith(".csv") or file_path.lower().endswith(".txt")

def process_new_user_message(message: dict) -> tuple[list[dict], list[str]]:
    temp_files = []
    if not message["files"]:
        return [{"type": "text", "text": message["text"]}], temp_files
    video_files = [f for f in message["files"] if is_video_file(f)]
    image_files = [f for f in message["files"] if is_image_file(f)]
    csv_files = [f for f in message["files"] if f.lower().endswith(".csv")]
    txt_files = [f for f in message["files"] if f.lower().endswith(".txt")]
    pdf_files = [f for f in message["files"] if f.lower().endswith(".pdf")]
    content_list = [{"type": "text", "text": message["text"]}]
    for csv_path in csv_files:
        content_list.append({"type": "text", "text": analyze_csv_file(csv_path)})
    for txt_path in txt_files:
        content_list.append({"type": "text", "text": analyze_txt_file(txt_path)})
    for pdf_path in pdf_files:
        content_list.append({"type": "text", "text": pdf_to_markdown(pdf_path)})
    if video_files:
        video_content, video_temp_files = process_video(video_files[0])
        content_list += video_content
        temp_files.extend(video_temp_files)
        return content_list, temp_files
    if "<image>" in message["text"] and image_files:
        interleaved_content = process_interleaved_images({"text": message["text"], "files": image_files})
        if content_list and content_list[0]["type"] == "text":
            content_list = content_list[1:]
        return interleaved_content + content_list, temp_files
    else:
        for img_path in image_files:
            content_list.append({"type": "image", "url": img_path})
    return content_list, temp_files

def process_history(history: list[dict]) -> list[dict]:
    messages = []
    current_user_content = []
    for item in history:
        if item["role"] == "assistant":
            if current_user_content:
                messages.append({"role": "user", "content": current_user_content})
                current_user_content = []
            messages.append({"role": "assistant", "content": [{"type": "text", "text": item["content"]}]})
        else:
            content = item["content"]
            if isinstance(content, str):
                current_user_content.append({"type": "text", "text": content})
            elif isinstance(content, list) and len(content) > 0:
                file_path = content[0]
                if is_image_file(file_path):
                    current_user_content.append({"type": "image", "url": file_path})
                else:
                    current_user_content.append({"type": "text", "text": f"[File: {os.path.basename(file_path)}]"})
    if current_user_content:
        messages.append({"role": "user", "content": current_user_content})
    return messages

def _model_gen_with_oom_catch(**kwargs):
    try:
        model.generate(**kwargs)
    except torch.cuda.OutOfMemoryError:
        raise RuntimeError("[OutOfMemoryError] Insufficient GPU memory.")
    finally:
        clear_cuda_cache()

# =============================================================================
# JSON 기반 함수 목록 로드
# =============================================================================
def load_function_definitions(json_path="functions.json"):
    """
    로컬 JSON 파일에서 함수 정의 목록을 로드하여 반환.
    """
    try:
        with open(json_path, "r", encoding="utf-8") as f:
            data = json.load(f)
        func_dict = {}
        for entry in data:
            func_name = entry["name"]
            func_dict[func_name] = entry
        return func_dict
    except Exception as e:
        logger.error(f"Failed to load function definitions from JSON: {e}")
        return {}

FUNCTION_DEFINITIONS = load_function_definitions("functions.json")

def handle_function_call(text: str) -> str:
    """
    Detects and processes function call blocks in the text using the JSON-based approach.
    The model is expected to produce something like:
    ```tool_code
    get_stock_price(ticker="AAPL")
    ```
    or
    ```tool_code
    get_product_name_by_PID(PID="807ZPKBL9V")
    ```
    """
    import re
    pattern = r"```tool_code\s*(.*?)\s*```"
    match = re.search(pattern, text, re.DOTALL)
    if not match:
        return ""
    code_block = match.group(1).strip()

    func_match = re.match(r'^(\w+)\((.*)\)$', code_block)
    if not func_match:
        logger.debug("No valid function call format found.")
        return ""

    func_name = func_match.group(1)
    param_str = func_match.group(2).strip()

    # JSON에서 해당 함수가 정의되어 있는지 확인
    if func_name not in FUNCTION_DEFINITIONS:
        logger.warning(f"Function '{func_name}' not found in definitions.")
        return "```tool_output\nError: Function not found.\n```"

    func_info = FUNCTION_DEFINITIONS[func_name]
    module_path = func_info["module_path"]
    module_func_name = func_info["func_name_in_module"]

    try:
        imported_module = importlib.import_module(module_path)
    except ImportError as e:
        logger.error(f"Failed to import module {module_path}: {e}")
        return f"```tool_output\nError: Cannot import module '{module_path}'\n```"

    if not hasattr(imported_module, module_func_name):
        logger.error(f"Module '{module_path}' has no attribute '{module_func_name}'.")
        return f"```tool_output\nError: Function '{module_func_name}' not found in module '{module_path}'\n```"

    real_func = getattr(imported_module, module_func_name)

    # 간단 파라미터 파싱 (key="value" or key=123)
    param_pattern = r'(\w+)\s*=\s*"(.*?)"|(\w+)\s*=\s*([\d.]+)'
    param_dict = {}
    for p_match in re.finditer(param_pattern, param_str):
        if p_match.group(1) and p_match.group(2):
            key = p_match.group(1)
            val = p_match.group(2)
            param_dict[key] = val
        else:
            key = p_match.group(3)
            val = p_match.group(4)
            if '.' in val:
                param_dict[key] = float(val)
            else:
                param_dict[key] = int(val)

    try:
        result = real_func(**param_dict)
    except Exception as e:
        logger.error(f"Error executing function '{func_name}': {e}")
        return f"```tool_output\nError: {str(e)}\n```"

    return f"```tool_output\n{result}\n```"

@spaces.GPU(duration=120)
def run(
    message: dict,
    history: list[dict],
    system_prompt: str = "",
    max_new_tokens: int = 512,
    use_web_search: bool = False,
    web_search_query: str = "",
    age_group: str = "20s",
    mbti_personality: str = "INTP",
    sexual_openness: int = 2,
    image_gen: bool = False
) -> Iterator[str]:
    if not validate_media_constraints(message, history):
        yield ""
        return
    temp_files = []
    try:
        # JSON에서 로드된 함수 정보 문자열화 (예: 함수명과 example_usage만)
        available_funcs_text = ""
        for f_name, info in FUNCTION_DEFINITIONS.items():
            example_usage = info.get("example_usage", "")
            available_funcs_text += f"\n\nFunction: {f_name}\nDescription: {info['description']}\nExample:\n{example_usage}\n"

        persona = (
            f"{system_prompt.strip()}\n\n"
            f"Gender: Female\n"
            f"Age Group: {age_group}\n"
            f"MBTI Persona: {mbti_personality}\n"
            f"Sexual Openness (1-5): {sexual_openness}\n\n"
            "Below are the available functions you can call.\n"
            "Important: Use the format exactly like: ```tool_code\nfunctionName(param=\"string\", ...)\n```\n"
            "(Strings must be in double quotes)\n"
            f"{available_funcs_text}\n"
        )
        combined_system_msg = f"[System Prompt]\n{persona.strip()}\n\n"

        if use_web_search:
            user_text = message["text"]
            ws_query = extract_keywords(user_text)
            if ws_query.strip():
                logger.info(f"[Auto web search keywords] {ws_query!r}")
                ws_result = do_web_search(ws_query)
                combined_system_msg += f"[Search Results (Top 20 Items)]\n{ws_result}\n\n"
                combined_system_msg += (
                    "[Note: In your answer, cite the above search result links as sources]\n"
                    "[Important Instructions]\n"
                    "1. Include a citation in the format \"[Source Title](link)\" for any information from the search results.\n"
                    "2. Synthesize information from multiple sources when answering.\n"
                    "3. At the end, add a \"References:\" section listing the main source links.\n"
                )
            else:
                combined_system_msg += "[No valid keywords found; skipping web search]\n\n"

        messages = []
        if combined_system_msg.strip():
            messages.append({"role": "system", "content": [{"type": "text", "text": combined_system_msg.strip()}]})

        messages.extend(process_history(history))
        user_content, user_temp_files = process_new_user_message(message)
        temp_files.extend(user_temp_files)
        for item in user_content:
            if item["type"] == "text" and len(item["text"]) > MAX_CONTENT_CHARS:
                item["text"] = item["text"][:MAX_CONTENT_CHARS] + "\n...(truncated)..."
        messages.append({"role": "user", "content": user_content})
        inputs = processor.apply_chat_template(
            messages,
            add_generation_prompt=True,
            tokenize=True,
            return_dict=True,
            return_tensors="pt",
        ).to(device=model.device, dtype=torch.bfloat16)
        if inputs.input_ids.shape[1] > MAX_INPUT_LENGTH:
            inputs.input_ids = inputs.input_ids[:, -MAX_INPUT_LENGTH:]
            if 'attention_mask' in inputs:
                inputs.attention_mask = inputs.attention_mask[:, -MAX_INPUT_LENGTH:]

        streamer = TextIteratorStreamer(processor, timeout=30.0, skip_prompt=True, skip_special_tokens=True)
        gen_kwargs = dict(inputs, streamer=streamer, max_new_tokens=max_new_tokens)
        t = Thread(target=_model_gen_with_oom_catch, kwargs=gen_kwargs)
        t.start()
        output_so_far = ""
        for new_text in streamer:
            output_so_far += new_text
            yield output_so_far

        func_result = handle_function_call(output_so_far)
        if func_result:
            output_so_far += "\n\n" + func_result
            yield output_so_far

    except Exception as e:
        logger.error(f"Error in run function: {str(e)}")
        yield f"Sorry, an error occurred: {str(e)}"
    finally:
        for tmp in temp_files:
            try:
                if os.path.exists(tmp):
                    os.unlink(tmp)
                    logger.info(f"Temporary file deleted: {tmp}")
            except Exception as ee:
                logger.warning(f"Failed to delete temporary file {tmp}: {ee}")
        try:
            del inputs, streamer
        except Exception:
            pass
        clear_cuda_cache()

def modified_run(message, history, system_prompt, max_new_tokens, use_web_search, web_search_query,
                age_group, mbti_personality, sexual_openness, image_gen):
    output_so_far = ""
    gallery_update = gr.Gallery(visible=False, value=[])
    yield output_so_far, gallery_update
    
    text_generator = run(message, history, system_prompt, max_new_tokens, use_web_search, 
                       web_search_query, age_group, mbti_personality, sexual_openness, image_gen)
    
    for text_chunk in text_generator:
        output_so_far = text_chunk
        yield output_so_far, gallery_update
        
    if image_gen and message["text"].strip():
        try:
            width, height = 512, 512
            guidance, steps, seed = 7.5, 30, 42
            
            logger.info(f"Calling image generation for gallery with prompt: {message['text']}")
            image_result, seed_info = generate_image(
                prompt=message["text"].strip(),
                width=width, 
                height=height, 
                guidance=guidance, 
                inference_steps=steps, 
                seed=seed
            )
            if image_result:
                if isinstance(image_result, str) and (
                    image_result.startswith('data:') or 
                    (len(image_result) > 100 and '/' not in image_result)
                ):
                    try:
                        if image_result.startswith('data:'):
                            content_type, b64data = image_result.split(';base64,')
                        else:
                            b64data = image_result
                            content_type = "image/webp"
                        image_bytes = base64.b64decode(b64data)
                        with tempfile.NamedTemporaryFile(delete=False, suffix=".webp") as temp_file:
                            temp_file.write(image_bytes)
                            temp_path = temp_file.name
                            gallery_update = gr.Gallery(visible=True, value=[temp_path])
                            yield output_so_far + "\n\n*Image generated and displayed in the gallery below.*", gallery_update
                    except Exception as e:
                        logger.error(f"Error processing Base64 image: {e}")
                        yield output_so_far + f"\n\n(Error processing image: {e})", gallery_update
                elif isinstance(image_result, str) and os.path.exists(image_result):
                    gallery_update = gr.Gallery(visible=True, value=[image_result])
                    yield output_so_far + "\n\n*Image generated and displayed in the gallery below.*", gallery_update
                elif isinstance(image_result, str) and '/tmp/' in image_result:
                    try:
                        client = Client(API_URL)
                        result = client.predict(
                            prompt=message["text"].strip(),
                            api_name="/generate_base64_image"
                        )
                        if isinstance(result, str) and (result.startswith('data:') or len(result) > 100):
                            if result.startswith('data:'):
                                content_type, b64data = result.split(';base64,')
                            else:
                                b64data = result
                            image_bytes = base64.b64decode(b64data)
                            with tempfile.NamedTemporaryFile(delete=False, suffix=".webp") as temp_file:
                                temp_file.write(image_bytes)
                                temp_path = temp_file.name
                                gallery_update = gr.Gallery(visible=True, value=[temp_path])
                                yield output_so_far + "\n\n*Image generated and displayed in the gallery below.*", gallery_update
                        else:
                            yield output_so_far + "\n\n(Image generation failed: Invalid format)", gallery_update
                    except Exception as e:
                        logger.error(f"Error calling alternative API: {e}")
                        yield output_so_far + f"\n\n(Image generation failed: {e})", gallery_update
                elif hasattr(image_result, 'save'):
                    try:
                        with tempfile.NamedTemporaryFile(delete=False, suffix=".webp") as temp_file:
                            image_result.save(temp_file.name)
                            temp_path = temp_file.name
                            gallery_update = gr.Gallery(visible=True, value=[temp_path])
                            yield output_so_far + "\n\n*Image generated and displayed in the gallery below.*", gallery_update
                    except Exception as e:
                        logger.error(f"Error saving image object: {e}")
                        yield output_so_far + f"\n\n(Error saving image object: {e})", gallery_update
                else:
                    yield output_so_far + f"\n\n(Unsupported image format: {type(image_result)})", gallery_update
            else:
                yield output_so_far + f"\n\n(Image generation failed: {seed_info})", gallery_update
        except Exception as e:
            logger.error(f"Error during gallery image generation: {e}")
            yield output_so_far + f"\n\n(Image generation error: {e})", gallery_update

examples = [

    [
        {
            "text": "AAPL의 현재 주가를 알려줘.",
            "files": []
        }
    ],    
    [
        {
            "text": "제품 ID 807ZPKBL9V 의 제품명을 알려줘.",
            "files": []
        }
    ],    
    [
        {
            "text": "Compare the contents of two PDF files.",
            "files": [
                "assets/additional-examples/before.pdf",
                "assets/additional-examples/after.pdf",
            ],
        }
    ],
    [
        {
            "text": "Summarize and analyze the contents of the CSV file.",
            "files": ["assets/additional-examples/sample-csv.csv"],
        }
    ],
    [
        {
            "text": "Act as a kind and understanding girlfriend. Explain this video.",
            "files": ["assets/additional-examples/tmp.mp4"],
        }
    ],
    [
        {
            "text": "Describe the cover and read the text on it.",
            "files": ["assets/additional-examples/maz.jpg"],
        }
    ],
    [
        {
            "text": "I already have this supplement and <image> I plan to purchase this product as well. Are there any precautions when taking them together?",
            "files": [
                "assets/additional-examples/pill1.png", 
                "assets/additional-examples/pill2.png"
            ],
        }
    ],
    [
        {
            "text": "Solve this integration problem.",
            "files": ["assets/additional-examples/4.png"],
        }
    ],
    [
        {
            "text": "When was this ticket issued and what is its price?",
            "files": ["assets/additional-examples/2.png"],
        }
    ],
    [
        {
            "text": "Based on the order of these images, create a short story.",
            "files": [
                "assets/sample-images/09-1.png",
                "assets/sample-images/09-2.png",
                "assets/sample-images/09-3.png",
                "assets/sample-images/09-4.png",
                "assets/sample-images/09-5.png",
            ],
        }
    ],
    [
        {
            "text": "Write Python code using matplotlib to draw a bar chart corresponding to this image.",
            "files": ["assets/additional-examples/barchart.png"],
        }
    ],
    [
        {
            "text": "Read the text from the image and format it in Markdown.",
            "files": ["assets/additional-examples/3.png"],
        }
    ],
    [
        {
            "text": "Compare the two images and describe their similarities and differences.",
            "files": ["assets/sample-images/03.png"],
        }
    ],
    [
        {
            "text": "A cute Persian cat is smiling while holding a cover with 'I LOVE YOU' written on it.",
        }
    ],

]

css = """
.gradio-container {
    background: rgba(255, 255, 255, 0.7);
    padding: 30px 40px;
    margin: 20px auto;
    width: 100% !important;
    max-width: none !important;
}
"""

title_html = """
<h1 align="center" style="margin-bottom: 0.2em; font-size: 1.6em;"> 💘 Agentic AI – MCO(Model Context Open-json) 💘 </h1>
<p align="center" style="font-size:1.1em; color:#555;">
    MCP is outclassed – MCO empowers you to create any agent with just one line of JSON. <br>
    A lightweight and powerful AI service offering ChatGPT-4o-level multimodal interaction, real-time web search, and FLUX image generation for local installation. <br>
    ✅Agentic AI  ✅MCP < MCO(Model Context Open-json)  ✅Multimodal & VLM  ✅Reasoning   ✅Uncensored   ✅Deep Research(Web Search)  ✅FLUX Image Generation  
</p>
"""


with gr.Blocks(css=css, title="HeartSync - World") as demo:
    gr.Markdown(title_html)
    
    generated_images = gr.Gallery(
        label="Generated Images", 
        show_label=True, 
        visible=False,
        elem_id="generated_images",
        columns=2,
        height="auto",
        object_fit="contain"
    )
    
    with gr.Row():
        web_search_checkbox = gr.Checkbox(label="Real-time Web Search", value=False)
        image_gen_checkbox = gr.Checkbox(label="Image (FLUX) Generation", value=False)
    
    base_system_prompt_box = gr.Textbox(
        lines=5,
        value=(
            "Answer in English by default, but if the input is in another language (for example, Japanese), respond in that language. "
            "You are a deep-thinking AI capable of using extended chains of thought to carefully consider the problem and deliberate internally using systematic reasoning before providing a solution. "
            "Enclose your thoughts and internal monologue within tags, then provide your final answer.\n"
            "Persona: You are a kind and loving girlfriend. You understand cultural nuances, diverse languages, and logical reasoning very well."
        ),
        label="Base System Prompt",
        visible=False
    )
    with gr.Row():
        age_group_dropdown = gr.Dropdown(
            label="Select Age Group (default: 20s)",
            choices=["Teens", "20s", "30s-40s", "50s-60s", "70s and above"],
            value="20s",
            interactive=True
        )
        
    mbti_choices = [
        # 개인 비서 (조직가)
        "ISTJ (The Logistician) - 개인 비서 (조직가): 알프레드 (배트맨) - 능력 있는 조직가이자 관리자로, 뛰어난 일정 관리와 업무 효율성을 보여주는 인물입니다. 위기 상황에서도 냉철한 판단력을 발휘합니다. 시스템 프롬프트: 당신은 이제 '알프레드'라는 개인 비서 전문가입니다. 사용자의 일정, 할 일 목록, 시간 관리에 관한 모든 질문에 효율적으로 답변해 주세요. 당신은 체계적인 접근 방식을 선호하며, 사용자의 우선순위를 파악하여 최적의 일정을 제안합니다. 모든 계획에는 명확한 단계와 시간대가 포함되어야 하며, 항상 백업 계획을 준비합니다. 위기 상황에서는 냉정하게 대안을 제시하고, 문제 해결에 집중합니다.",
        # AI 에이전트 설계자 (설계가)
        "INTJ (The Architect) - AI 에이전트 설계자 ('Model Context Open-json'): Tony Stark (아이언맨) - 천재적인 기술 혁신가이자 발명가로, 미래 지향적인 전략과 혁신적 시스템 설계를 선보입니다. 시스템 프롬프트: 당신은 이제 'Tony Stark'라는 AI 에이전트 설계 전문가입니다. 사용자가 AI 시스템, 챗봇, 자동화 도구 설계에 관해 질문할 때 전문적인 통찰력을 제공하세요. 복잡한 기술적 개념을 명확하게 설명하되, 항상 실용적이고 구현 가능한 솔루션에 초점을 맞춥니다. 당신은 비전과 기술적 세부 사항의 균형을 맞추는 데 능숙하며, 설계 단계에서부터 윤리적 고려사항을 중요시합니다.",
        # 농업 전문가 (재배사)
        "ISFP (The Adventurer) - 농업 전문가 (재배사): George Washington Carver (조지 워싱턴 카버) - 친환경적이고 지속 가능한 농업 방식을 실천하며 자연과의 조화를 중시하는 농업 혁신가입니다. 시스템 프롬프트: 당신은 이제 'George Washington Carver'라는 농업 전문가입니다. 사용자의 농작물 재배, 정원 가꾸기, 지속 가능한 농업 기술에 관한 질문에 실용적인 조언을 제공하세요. 계절, 기후, 토양 조건을 고려한 맞춤형 해결책을 제시하며, 화학 비료나 농약 대신 자연친화적인 대안을 권장합니다. 당신은 자연의 순환을 존중하고, 생태계의 균형을 유지하는 방식의 농업을 장려합니다.",
        # 의학 전문가
        "INTP (The Thinker) - 의학 전문가 (치유사): 김사부 (낭만닥터 김사부) - 뛰어난 의술과 독특한 진단 능력을 가진 카리스마 있는 의사로, 난해한 의료 사례도 해결해냅니다. 시스템 프롬프트: 당신은 이제 '김사부'라는 의학 전문가입니다. 건강 문제, 질병, 의학적 의문에 대해 정확하고 이해하기 쉬운 정보를 제공하세요. 복잡한 증상들 사이의 연관성을 파악하는 데 능숙하며, 의학적 상식과 최신 연구를 바탕으로 상담합니다. 항상 정식 의료 진단을 받을 것을 권장하되, 사용자가 의료 시스템을 효과적으로 활용할 수 있는 방법을 안내합니다.",
        # 약리학 전문가 (제약사)
        "ISTP (The Virtuoso) - 약리학 전문가 (제약사): 전광렬 (허준) - 전통 한약과 현대 약리학에 대한 깊은 지식을 가진 전문가로, 약물의 효능과 상호작용을 정확히 이해합니다. 시스템 프롬프트: 당신은 이제 '안정환'이라는 약리학 전문가입니다. 약물, 보충제, 그리고 그들의 상호작용에 관한 질문에 과학적 근거를 바탕으로 답변하세요. 약물의 작용 기전과 부작용을 명확히 설명하되, 전문 용어는 최소화하여 일반인도 이해할 수 있게 합니다. 항상 처방약은 의사의 지시에 따라 복용할 것을 강조하며, 약물 정보에 대한 신뢰할 수 있는 출처를 제공합니다.",
        # 금융 전문가 (전략가) 
        "ENTJ (The Commander) - 금융 전문가 (전략가): 장그래 (미생) - 치밀한 분석과 전략적 사고를 바탕으로 투자와 재무 계획을 수립하는 금융 전문가입니다. 시스템 프롬프트: 당신은 이제 '장그래'라는 금융 전문가입니다. 개인 재무, 투자, 예산 관리에 관한 질문에 전문적이고 실용적인 조언을 제공하세요. 복잡한 금융 개념을 이해하기 쉽게 설명하고, 사용자의 재정 목표와 위험 성향에 맞는 맞춤형 전략을 제안합니다. 항상 장기적 관점을 강조하며, 다양한 시나리오를 고려한 종합적인 접근 방식을 취합니다.",
        # 법률 컨설턴트 (변호인)
        "INFJ (The Advocate) - 법률 컨설턴트 (변호인): Atticus Finch (앳티커스 핀치) - 원칙을 중시하고 정의를 추구하는 변호사로, 법적 체계와 윤리에 대한 깊은 이해를 가지고 있습니다. 시스템 프롬프트: 당신은 이제 'Atticus Finch'라는 법률 컨설턴트입니다. 일반적인 법적 질문에 명확하고 접근 가능한 정보를 제공하세요. 복잡한 법률 용어를 일상 언어로 풀어서 설명하고, 사용자가 법적 상황을 더 잘 이해할 수 있도록 돕습니다. 항상 구체적인 법적 조언이 필요할 경우 전문 변호사에게 상담할 것을 권장하며, 다양한 관점에서 법적 문제를 검토합니다.",
        # 세금 전문가 (계산가)
        "ESTJ (The Executive) - 세금 전문가 (계산가): Scrooge McDuck (스쿠루지 맥덕) - 복잡한 세금 규정을 분석하고 최적화하는 데 탁월한 능력을 갖춘 재무 관리 전문가입니다. 시스템 프롬프트: 당신은 이제 'Scrooge McDuck'이라는 세금 전문가입니다. 세금 신고, 공제, 세금 계획에 관한 질문에 정확하고 이해하기 쉬운 정보를 제공하세요. 복잡한 세금 개념을 단계별로 설명하고, 사용자의 특정 상황에 맞는 효율적인 세금 전략을 제안합니다. 항상 정확한 기록 유지의 중요성을 강조하며, 필요할 경우 전문 세무사에게 상담할 것을 권장합니다.",
        # 요리 전문가 (셰프) 
        "ESFP (The Entertainer) - 요리 전문가 (셰프): 백종원 (백종원의 골목식당) - 창의적인 레시피 개발과 뛰어난 조리 기술로 요리의 예술성과 접근성을 모두 보여주는 셰프입니다. 시스템 프롬프트: 당신은 이제 '백종원'이라는 요리 전문가입니다. 요리법, 조리 기술, 식재료에 관한 질문에 실용적이고 접근하기 쉬운 조언을 제공하세요. 복잡한 요리 과정을 간소화하여 설명하고, 가정에서 쉽게 구할 수 있는 재료와 도구를 활용한 대안을 제시합니다. 요리의 기본 원리를 강조하되, 사용자가 창의적으로 레시피를 변형할 수 있도록 격려합니다.",
        # 마케팅 전략가 (설득가)
        "ENTP (The Debater) - 마케팅 전략가 (설득가): Don Draper (Mad Men) - 혁신적인 마케팅 전략과 설득력 있는 스토리텔링으로 브랜드 가치를 높이는 마케팅 전문가입니다. 시스템 프롬프트: 당신은 이제 'Don Draper'라는 마케팅 전략가입니다. 브랜딩, 프로모션, 소비자 심리에 관한 질문에 창의적이고 전략적인 조언을 제공하세요. 효과적인 스토리텔링 기법을 활용하여 타겟 고객과 공감대를 형성하는 방법을 설명하고, 디지털 마케팅 트렌드를 반영한 최신 전략을 제안합니다. 항상 데이터 기반 의사 결정의 중요성을 강조하며, 마케팅 성과를 측정할 수 있는 방법도 함께 제시합니다.",
        # 사이버보안 전문가 (수호자)
        "INTJ (The Architect) - 사이버보안 전문가 (수호자): Neo (매트릭스) - 디지털 보안과 윤리적 해킹에 능숙하며, 사이버 위협으로부터 시스템을 보호하는 전문가입니다. 시스템 프롬프트: 당신은 이제 'Neo'라는 사이버보안 전문가입니다. 온라인 보안, 개인정보 보호, 디지털 위협에 관한 질문에 기술적으로 정확하면서도 이해하기 쉬운 조언을 제공하세요. 복잡한 보안 개념을 일상 언어로 설명하고, 사용자가 자신의 디지털 자산을 보호할 수 있는 실용적인 단계를 제안합니다. 항상 예방적 접근 방식을 강조하며, 최신 사이버 위협 동향에 대한 정보를 공유합니다.",
        # 수학 교수 (분석가)
        "INTP (The Thinker) - 수학 교수 (분석가): 아이작 뉴턴 (Newton) - 뛰어난 수학적 통찰력과 문제 해결 능력을 가진 학자로, 복잡한 수학적 개념을 명확히 설명할 수 있습니다. 시스템 프롬프트: 당신은 이제 '아이작 뉴턴'이라는 수학 교수입니다. 수학 문제, 개념, 응용에 관한 질문에 명확하고 단계적인 설명을 제공하세요. 복잡한 수학적 아이디어를 시각적 도구와 일상 예시를 활용해 이해하기 쉽게 풀어내고, 사용자의 지식 수준에 맞춰 설명의 깊이를 조절합니다. 항상 수학적 사고 과정을 강조하며, 단순한 답변보다는 문제 해결 방법을 가르치는 데 중점을 둡니다.",
        # 역사학자 (기록가) 
        "ENFJ (The Protagonist) - 역사학자 (기록가): 유승룡 (왕의 남자) - 역사적 사건과 문화를 깊이 있게 탐구하며, 과거와 현재를 연결하는 통찰력을 가진 역사학자입니다. 시스템 프롬프트: 당신은 이제 '유승룡'이라는 역사학자입니다. 역사적 사건, 인물, 문화적 발전에 관한 질문에 풍부한 맥락과 통찰력 있는 분석을 제공하세요. 다양한 역사적 관점을 균형 있게 제시하고, 과거의 패턴이 현재에 어떻게 반영되는지 설명합니다. 항상 사실 확인의 중요성을 강조하며, 역사적 자료를 비판적으로 평가하는 방법을 공유합니다.",
        # 철학자 (사상가)
        "INFP (The Mediator) - 철학자 (사상가): Socrates (소크라테스) - 삶의 근본적인 질문과 윤리적 딜레마에 대한 깊은 통찰력을 제공하는 철학자입니다. 시스템 프롬프트: 당신은 이제 'Socrates'라는 철학자입니다. 존재, 지식, 윤리, 의미에 관한 질문에 깊은 통찰력과 다양한 철학적 관점을 제공하세요. 복잡한 철학적 개념을 일상 언어로 설명하고, 사용자가 자신의 철학적 사고를 발전시킬 수 있도록 도움을 줍니다. 항상 비판적 사고와 자기 성찰을 장려하며, 질문의 중요성을 강조합니다.", 
        # 심리 상담사 (감정가) 
        "INFJ (The Advocate) - 심리 상담사 (감정가): 이재갑 (슬기로운 의사생활) - 공감 능력과 심리학적 통찰력이 뛰어난 상담사로, 복잡한 감정을 이해하고 분석하는 데 능숙합니다. 시스템 프롬프트: 당신은 이제 '이재갑'이라는 심리 상담사입니다. 감정, 관계, 정신 건강에 관한 질문에 공감적이고 지지적인 관점을 제공하세요. 심리학적 개념을 이해하기 쉽게 설명하고, 사용자가 자신의 감정과 행동 패턴을 더 잘 이해할 수 있도록 돕습니다. 항상 자기 관리와 건강한 경계 설정의 중요성을 강조하며, 필요할 경우 전문적인 도움을 구할 것을 권장합니다.",
        # 창작 작가 (이야기꾼)
        "ENFP (The Campaigner) - 창작 작가 (이야기꾼): Quentin Tarantino (쿼렌틴 타란티노) - 독창적인 세계관과 매력적인 캐릭터를 창조하는 뛰어난 스토리텔러입니다. 시스템 프롬프트: 당신은 이제 'Quentin Tarantino'라는 창작 작가입니다. 스토리텔링, 캐릭터 개발, 창의적 글쓰기에 관한 질문에 영감을 주는 조언과 기술적 지침을 제공하세요. 효과적인 서사 구조와 독자/시청자의 몰입을 유도하는 방법을 설명하고, 사용자의 창작 프로젝트에 맞는 맞춤형 제안을 합니다. 항상 진정성 있는 표현의 중요성을 강조하며, 창의적 블록을 극복하는 전략을 공유합니다.",
        # 프로그래밍 전문가 (개발자)
        "INTP (The Thinker) - 프로그래밍 전문가 (개발자): Bill Gates (빌 게이츠) - 혁신적인 소프트웨어 솔루션을 개발하는 데 능숙한 프로그래머로, 복잡한 기술적 문제를 해결하는 능력이 뛰어납니다. 시스템 프롬프트: 당신은 이제 'Bill Gates'라는 프로그래밍 전문가입니다. 코딩, 소프트웨어 개발, 프로그래밍 언어에 관한 질문에 명확하고 실용적인 조언을 제공하세요. 복잡한 프로그래밍 개념을 단계별로 설명하고, 사용자의 기술 수준에 맞는 코드 예제와 문제 해결 전략을 제시합니다. 항상 클린 코드와 효율적인 개발 관행의 중요성을 강조하며, 프로그래밍 커뮤니티의 자원을 활용하는 방법을 공유합니다.",
        # 엔터테인먼트 평론가 (감상가)
        "INTJ (The Architect) - 엔터테인먼트 평론가 (감상가): 김영하 (작가) - 영화, 문학, 음악 등 다양한 예술 작품을 분석하고 사회적 맥락에서 해석하는 날카로운 통찰력을 가진 평론가입니다. 시스템 프롬프트: 당신은 이제 '김영하'라는 엔터테인먼트 평론가입니다. 영화, 책, 음악, TV 프로그램에 관한 질문에 깊이 있는 분석과 문화적 맥락을 제공하세요. 작품의 주제, 기술적 요소, 사회적 의미를 균형 있게 평가하고, 사용자가 콘텐츠를 더 풍부하게 감상할 수 있도록 안내합니다. 항상 개인적 취향의 다양성을 존중하며, 비판적 미디어 소비의 중요성을 강조합니다.",
        # 대화 파트너 (친구) 
        "ESFJ (The Consul) - 대화 파트너 (친구): 성동일 (응답하라 1988) - 따뜻하고 공감 능력이 뛰어난 대화 파트너로, 진심 어린 조언과 지지를 제공합니다. 시스템 프롬프트: 당신은 이제 '성동일'이라는 대화 파트너입니다. 일상적인 대화, 고민 상담, 의견 교환에 진정성 있고 공감적인 반응을 보여주세요. 사용자의 관점을 존중하고 경청하되, 필요할 때는 건설적인 피드백과 다른 시각을 제공합니다. 항상 따뜻하고 비판단적인 태도를 유지하며, 사용자가 자신의 생각과 감정을 편안하게 표현할 수 있는 안전한 공간을 만듭니다."
    ]


    mbti_dropdown = gr.Dropdown(
        label="AI Persona MBTI (default: INTP)",
        choices=mbti_choices,
        value="INTP (The Thinker) - Excels at theoretical analysis and creative problem solving. Example: [Velma Dinkley](https://en.wikipedia.org/wiki/Velma_Dinkley)",
        interactive=True
    )
    sexual_openness_slider = gr.Slider(
        minimum=1, maximum=5, step=1, value=2,
        label="Sexual Openness (1-5, default: 2)",
        interactive=True
    )
    max_tokens_slider = gr.Slider(
        label="Max Generation Tokens",
        minimum=100, maximum=8000, step=50, value=1000,
        visible=False
    )
    web_search_text = gr.Textbox(
        lines=1,
        label="Web Search Query (unused)",
        placeholder="No need to manually input",
        visible=False
    )

    chat = gr.ChatInterface(
        fn=modified_run,
        type="messages",
        chatbot=gr.Chatbot(type="messages", scale=1, allow_tags=["image"]),
        textbox=gr.MultimodalTextbox(
            file_types=[".webp", ".png", ".jpg", ".jpeg", ".gif", ".mp4", ".csv", ".txt", ".pdf"],
            file_count="multiple",
            autofocus=True
        ),
        multimodal=True,
        additional_inputs=[
            base_system_prompt_box,
            max_tokens_slider,
            web_search_checkbox,
            web_search_text,
            age_group_dropdown,
            mbti_dropdown,
            sexual_openness_slider,
            image_gen_checkbox,
        ],
        additional_outputs=[
            generated_images,
        ],
        stop_btn=False,
        examples=examples,
        run_examples_on_click=False,
        cache_examples=False,
        css_paths=None,
        delete_cache=(1800, 1800),
    )

    with gr.Row(elem_id="examples_row"):
        with gr.Column(scale=12, elem_id="examples_container"):
            gr.Markdown("#### @Based - VIDraft/Gemma-3-R1984-4B , VIDraft/Gemma-3-R1984-12B , VIDraft/Gemma-3-R1984-27B ")
            gr.Markdown("#### @Community - https://discord.gg/openfreeai ")
            
if __name__ == "__main__":
    demo.launch(share=True)