Blame for app.py

"""Configurable text-only Gradio chat UI for an OpenAI-compatible API.

Recommended requirements.txt:
    gradio==6.28.0
    openai==2.54.0

Required Space settings:
    MODEL_API_BASE or OPENAI_BASE_URL
    MODEL_API_KEY  or OPENAI_API_KEY (store as a Secret)

Model and runtime settings:
    MODEL_ID or MODEL_NAME, SYSTEM_PROMPT, MAX_OUTPUT_TOKENS, TEMPERATURE,
    TOP_P, API_TIMEOUT_SECONDS, API_MAX_RETRIES, CONCURRENCY_LIMIT,
    QUEUE_MAX_SIZE, STREAM_UPDATE_INTERVAL_SECONDS, SERVER_NAME, SERVER_PORT,
    SHARE, ANALYTICS_ENABLED

Brand and UI settings:
    APP_TITLE, APP_HEADING_MARKDOWN, MODEL_LABEL, CHAT_PLACEHOLDER,
    INPUT_PLACEHOLDER, SEND_BUTTON_TEXT, STOP_BUTTON_TEXT, CLEAR_BUTTON_TEXT,
    THINKING_TITLE, STATUS_TITLE, NOTICE_TITLE, STATUS_GENERATING_TEXT,
    STOPPED_TEXT, CHATBOT_HEIGHT, CHATBOT_LAYOUT, CHATBOT_BUTTONS,
    INPUT_LINES, INPUT_MAX_LINES

All validation errors, upstream errors, and generation notices can also be
overridden with the ERROR_* and NOTICE_* environment variables defined below.

The UI renders Markdown in both reasoning and final answers. Reasoning must be
provided by the API in `reasoning_content` or `reasoning`; this app does not
split <think> tags embedded in the normal `content` field. Reasoning and UI
notices are kept out of the model conversation history.
"""

from __future__ import annotations

import asyncio
import copy
import logging
import os
import time
import uuid

import gradio as gr
from openai import APIConnectionError, APIStatusError, APITimeoutError, AsyncOpenAI


def env_bool(name: str, default: bool) -> bool:
    raw = os.getenv(name)
    if raw is None:
        return default
    normalized = raw.strip().lower()
    if normalized in {"1", "true", "yes", "on"}:
        return True
    if normalized in {"0", "false", "no", "off"}:
        return False
    raise ValueError(f"{name} must be true or false.")


def env_int(name: str, default: int) -> int:
    raw = os.getenv(name)
    if raw is None:
        return default
    try:
        return int(raw)
    except ValueError as exc:
        raise ValueError(f"{name} must be an integer.") from exc


def env_float(name: str, default: float) -> float:
    raw = os.getenv(name)
    if raw is None:
        return default
    try:
        return float(raw)
    except ValueError as exc:
        raise ValueError(f"{name} must be a number.") from exc


def first_env(*names: str, default: str = "") -> str:
    for name in names:
        value = os.getenv(name)
        if value is not None and value.strip():
            return value.strip()
    return default


# Model API and generation settings.
MODEL_ID = first_env(
    "MODEL_ID", "MODEL_NAME", default="IQuest-Q1"
)
API_BASE = first_env("MODEL_API_BASE", "OPENAI_BASE_URL")
API_KEY = first_env("MODEL_API_KEY", "OPENAI_API_KEY")
SYSTEM_PROMPT = os.getenv(
    "SYSTEM_PROMPT",
    "You are IQuest-Q1, a helpful AI assistent developed by IQuestLab.",
)
MAX_OUTPUT_TOKENS = env_int("MAX_OUTPUT_TOKENS", 16384)
TEMPERATURE = env_float("TEMPERATURE", 1.0)
TOP_P = env_float("TOP_P", 0.95)
API_TIMEOUT_SECONDS = env_float("API_TIMEOUT_SECONDS", 180.0)
API_MAX_RETRIES = env_int("API_MAX_RETRIES", 0)
CONCURRENCY_LIMIT = env_int("CONCURRENCY_LIMIT", 10)
QUEUE_MAX_SIZE = env_int("QUEUE_MAX_SIZE", 32)
STREAM_UPDATE_INTERVAL_SECONDS = env_float("STREAM_UPDATE_INTERVAL_SECONDS", 0.05)

# Server settings.
SERVER_NAME = os.getenv("SERVER_NAME", "0.0.0.0")
SERVER_PORT = env_int("SERVER_PORT", 7860)
SHARE = env_bool("SHARE", False)
ANALYTICS_ENABLED = env_bool("ANALYTICS_ENABLED", False)
LOGGER_NAME = os.getenv("LOGGER_NAME", "iquest_chat")

# Brand, layout, and all user-visible UI text.
APP_TITLE = os.getenv("APP_TITLE", "IQuest-Q1")
APP_HEADING_MARKDOWN = os.getenv("APP_HEADING_MARKDOWN", f"# {APP_TITLE}")
MODEL_LABEL = os.getenv("MODEL_LABEL", MODEL_ID)
CHAT_PLACEHOLDER = os.getenv(
    "CHAT_PLACEHOLDER", "Enter a question or paste code to start a conversation."
)
INPUT_PLACEHOLDER = os.getenv(
    "INPUT_PLACEHOLDER", "Enter a question or paste code..."
)
SEND_BUTTON_TEXT = os.getenv("SEND_BUTTON_TEXT", "Send")
STOP_BUTTON_TEXT = os.getenv("STOP_BUTTON_TEXT", "Stop")
CLEAR_BUTTON_TEXT = os.getenv("CLEAR_BUTTON_TEXT", "Clear conversation")
THINKING_TITLE = os.getenv("THINKING_TITLE", "Thinking")
STATUS_TITLE = os.getenv("STATUS_TITLE", "Status")
NOTICE_TITLE = os.getenv("NOTICE_TITLE", "Notice")
STATUS_GENERATING_TEXT = os.getenv("STATUS_GENERATING_TEXT", "Generating...")
STOPPED_TEXT = os.getenv("STOPPED_TEXT", "Generation stopped.")
CHATBOT_HEIGHT = env_int("CHATBOT_HEIGHT", 560)
CHATBOT_LAYOUT = os.getenv("CHATBOT_LAYOUT", "panel")
CHATBOT_BUTTONS = [
    item.strip()
    for item in os.getenv("CHATBOT_BUTTONS", "copy").split(",")
    if item.strip()
]
INPUT_LINES = env_int("INPUT_LINES", 2)
INPUT_MAX_LINES = env_int("INPUT_MAX_LINES", 12)

# User-visible validation errors.
ERROR_EMPTY_INPUT = os.getenv("ERROR_EMPTY_INPUT", "Please enter a message.")
ERROR_ALREADY_RUNNING = os.getenv(
    "ERROR_ALREADY_RUNNING",
    "Stop the current response before sending another message.",
)
ERROR_MISSING_API_CONFIG = os.getenv(
    "ERROR_MISSING_API_CONFIG",
    "The model API endpoint or API key is not configured. Please contact the administrator.",
)

# User-visible upstream and runtime errors.
ERROR_TIMEOUT = os.getenv(
    "ERROR_TIMEOUT", "The model response timed out. Please try again."
)
ERROR_CONNECTION = os.getenv(
    "ERROR_CONNECTION", "Unable to connect to the model service. Please try again later."
)
ERROR_AUTH = os.getenv(
    "ERROR_AUTH",
    "Model service authentication failed. Please contact the administrator.",
)
ERROR_RATE_LIMIT = os.getenv(
    "ERROR_RATE_LIMIT", "The model service is busy or the quota is exhausted. Please try again later."
)
ERROR_BAD_REQUEST = os.getenv(
    "ERROR_BAD_REQUEST",
    "The model service rejected the request. The conversation may be too long or a request parameter may be unsupported.",
)
ERROR_HTTP = os.getenv(
    "ERROR_HTTP", "The model service returned HTTP {status_code}. Please try again later."
)
ERROR_GENERIC = os.getenv(
    "ERROR_GENERIC", "An unexpected error occurred while generating the response. Please try again."
)

# User-visible generation notices.
NOTICE_LENGTH_LIMIT = os.getenv(
    "NOTICE_LENGTH_LIMIT", "The output token limit was reached. The response may be incomplete."
)
NOTICE_CONTENT_FILTER = os.getenv(
    "NOTICE_CONTENT_FILTER", "The model service did not return the complete response."
)
NOTICE_EMPTY_RESPONSE = os.getenv(
    "NOTICE_EMPTY_RESPONSE",
    "The model returned no final answer. Please retry or check the output limit and API response format.",
)
NOTICE_MISSING_FINISH_REASON = os.getenv(
    "NOTICE_MISSING_FINISH_REASON",
    "No normal completion marker was received. The response may be incomplete.",
)

# Explicit Markdown and safety behavior for both Thinking and final answers.
RENDER_MARKDOWN = env_bool("RENDER_MARKDOWN", True)
LINE_BREAKS = env_bool("LINE_BREAKS", True)
SANITIZE_HTML = env_bool("SANITIZE_HTML", True)
ALLOW_TAGS = env_bool("ALLOW_TAGS", False)
AUTOSCROLL = env_bool("AUTOSCROLL", True)

logger = logging.getLogger(LOGGER_NAME)


def validate_config() -> None:
    if MAX_OUTPUT_TOKENS < 1:
        raise ValueError("MAX_OUTPUT_TOKENS must be greater than 0.")
    if API_TIMEOUT_SECONDS <= 0:
        raise ValueError("API_TIMEOUT_SECONDS must be greater than 0.")
    if API_MAX_RETRIES < 0:
        raise ValueError("API_MAX_RETRIES must be 0 or greater.")
    if not 0 <= TEMPERATURE <= 2:
        raise ValueError("TEMPERATURE must be between 0 and 2.")
    if not 0 < TOP_P <= 1:
        raise ValueError("TOP_P must be greater than 0 and no greater than 1.")
    if CONCURRENCY_LIMIT < 1:
        raise ValueError("CONCURRENCY_LIMIT must be greater than 0.")
    if QUEUE_MAX_SIZE < 1:
        raise ValueError("QUEUE_MAX_SIZE must be greater than 0.")
    if STREAM_UPDATE_INTERVAL_SECONDS < 0:
        raise ValueError("STREAM_UPDATE_INTERVAL_SECONDS must be 0 or greater.")
    if SERVER_PORT < 1 or SERVER_PORT > 65535:
        raise ValueError("SERVER_PORT must be between 1 and 65535.")
    if CHATBOT_HEIGHT < 1:
        raise ValueError("CHATBOT_HEIGHT must be greater than 0.")
    if CHATBOT_LAYOUT not in {"panel", "bubble"}:
        raise ValueError("CHATBOT_LAYOUT must be panel or bubble.")
    invalid_buttons = set(CHATBOT_BUTTONS) - {"copy", "copy_all", "share"}
    if invalid_buttons:
        raise ValueError(
            "CHATBOT_BUTTONS only supports copy, copy_all, and share."
        )
    if INPUT_LINES < 1 or INPUT_MAX_LINES < INPUT_LINES:
        raise ValueError("INPUT_MAX_LINES must be greater than or equal to INPUT_LINES.")


def new_session() -> dict:
    # Keep session state deepcopy-safe and isolated between browser sessions.
    return {"messages": [], "display": [], "run_id": None}


def ui_updates(state: dict, clear_input: bool = False) -> tuple:
    running = state["run_id"] is not None
    textbox_update = {"interactive": not running}
    if clear_input:
        textbox_update["value"] = ""
    return (
        copy.deepcopy(state["display"]),
        state,
        gr.update(**textbox_update),
        gr.update(visible=not running),
        gr.update(visible=running),
    )


async def submit(message: str, state: dict) -> tuple:
    """Validate and append a user message before the generation event starts."""
    if not isinstance(message, str) or not message.strip():
        raise gr.Error(ERROR_EMPTY_INPUT)
    if state["run_id"] is not None:
        raise gr.Error(ERROR_ALREADY_RUNNING)
    if not API_BASE or not API_KEY:
        raise gr.Error(ERROR_MISSING_API_CONFIG)

    # Preserve whitespace so pasted code keeps its indentation.
    user = {"role": "user", "content": message}
    state["messages"].append(user.copy())
    state["display"].append(user.copy())
    state["run_id"] = uuid.uuid4().hex
    job = {
        "id": state["run_id"],
        "messages": copy.deepcopy(state["messages"]),
        "display": copy.deepcopy(state["display"]),
    }
    state["display"].append(
        {
            "role": "assistant",
            "content": STATUS_GENERATING_TEXT,
            "metadata": {"title": STATUS_TITLE, "status": "pending"},
        }
    )
    return (*ui_updates(state, clear_input=True), job)


def publish(
    job: dict,
    state: dict,
    thinking: str,
    answer: str,
    *,
    done: bool = False,
    notice: str = "",
) -> tuple:
    """Build separate model and display histories for the current stream state."""
    state["messages"] = copy.deepcopy(job["messages"])
    state["display"] = copy.deepcopy(job["display"])
    if thinking:
        state["display"].append(
            {
                "role": "assistant",
                "content": thinking,
                "metadata": {
                    "title": THINKING_TITLE,
                    "status": "done" if done or answer else "pending",
                },
            }
        )
    if answer:
        reply = {"role": "assistant", "content": answer}
        state["messages"].append(reply.copy())
        state["display"].append(reply.copy())
    if notice:
        state["display"].append(
            {
                "role": "assistant",
                "content": notice,
                "metadata": {"title": NOTICE_TITLE, "status": "done"},
            }
        )
    return ui_updates(state)


def error_message(exc: Exception) -> str:
    if isinstance(exc, APITimeoutError):
        return ERROR_TIMEOUT
    if isinstance(exc, APIConnectionError):
        return ERROR_CONNECTION
    if isinstance(exc, APIStatusError):
        if exc.status_code in (401, 403):
            return ERROR_AUTH
        if exc.status_code == 429:
            return ERROR_RATE_LIMIT
        if exc.status_code in (400, 413, 422):
            return ERROR_BAD_REQUEST
        return ERROR_HTTP.replace("{status_code}", str(exc.status_code))
    return ERROR_GENERIC


async def respond(job: dict | None, state: dict):
    # Clear or Stop may invalidate a queued job before it starts.
    if not job or state["run_id"] != job["id"]:
        return

    messages = copy.deepcopy(job["messages"])
    if SYSTEM_PROMPT.strip():
        messages.insert(0, {"role": "system", "content": SYSTEM_PROMPT})

    thinking = ""
    answer = ""
    notice = ""
    finish_reason = None
    last_update = 0.0
    try:
        # A separate client per generation makes cancellation close its connection.
        async with AsyncOpenAI(
            base_url=API_BASE,
            api_key=API_KEY,
            timeout=API_TIMEOUT_SECONDS,
            max_retries=API_MAX_RETRIES,
        ) as client:
            async with await client.chat.completions.create(
                model=MODEL_ID,
                messages=messages,
                max_tokens=MAX_OUTPUT_TOKENS,
                temperature=TEMPERATURE,
                top_p=TOP_P,
                stream=True,
            ) as stream:
                async for chunk in stream:
                    if state["run_id"] != job["id"]:
                        return
                    if not chunk.choices:
                        # Some providers send a usage-only final chunk.
                        continue
                    choice = chunk.choices[0]
                    if choice.finish_reason:
                        finish_reason = choice.finish_reason
                    delta = choice.delta
                    if delta is None:
                        continue

                    thought = getattr(delta, "reasoning_content", None)
                    if not isinstance(thought, str):
                        thought = getattr(delta, "reasoning", None)
                    content = delta.content or getattr(delta, "refusal", None)

                    changed = False
                    if isinstance(thought, str) and thought:
                        thinking += thought
                        changed = True
                    if isinstance(content, str) and content:
                        answer += content
                        changed = True

                    now = time.monotonic()
                    if changed and now - last_update >= STREAM_UPDATE_INTERVAL_SECONDS:
                        yield publish(job, state, thinking, answer)
                        last_update = now

        if finish_reason == "length":
            notice = NOTICE_LENGTH_LIMIT
        elif finish_reason == "content_filter":
            notice = NOTICE_CONTENT_FILTER
        elif not answer.strip():
            notice = NOTICE_EMPTY_RESPONSE
        elif finish_reason is None:
            notice = NOTICE_MISSING_FINISH_REASON
    except asyncio.CancelledError:
        # Stop and Clear restore the UI; cancellation closes the HTTP stream.
        raise
    except Exception as exc:
        # Avoid logging API keys, conversation content, or upstream response bodies.
        logger.warning(
            "generation_failed type=%s http_status=%s",
            type(exc).__name__,
            getattr(exc, "status_code", None),
        )
        notice = error_message(exc)

    if state["run_id"] == job["id"]:
        state["run_id"] = None
        yield publish(job, state, thinking, answer, done=True, notice=notice)


async def stop(state: dict) -> tuple:
    if state["run_id"] is not None:
        state["run_id"] = None
        # Remove the temporary status message when stopping before the first token.
        state["display"] = [
            message
            for message in state["display"]
            if not (
                message.get("content") == STATUS_GENERATING_TEXT
                and message.get("metadata", {}).get("title") == STATUS_TITLE
                and message.get("metadata", {}).get("status") == "pending"
            )
        ]
        for message in state["display"]:
            if message.get("metadata"):
                message["metadata"]["status"] = "done"
        state["display"].append(
            {
                "role": "assistant",
                "content": STOPPED_TEXT,
                "metadata": {"title": NOTICE_TITLE, "status": "done"},
            }
        )
    return ui_updates(state)


async def clear(state: dict) -> tuple:
    # Mutate in place so any running task sees that its run ID is invalid.
    state.update(new_session())
    return ui_updates(state, clear_input=True)


def build_app() -> gr.Blocks:
    validate_config()
    with gr.Blocks(title=APP_TITLE, analytics_enabled=ANALYTICS_ENABLED) as demo:
        state = gr.State(value=new_session())
        job = gr.State(value=None)
        gr.Markdown(APP_HEADING_MARKDOWN)
        chatbot = gr.Chatbot(
            label=MODEL_LABEL,
            height=CHATBOT_HEIGHT,
            layout=CHATBOT_LAYOUT,
            render_markdown=RENDER_MARKDOWN,
            line_breaks=LINE_BREAKS,
            sanitize_html=SANITIZE_HTML,
            allow_tags=ALLOW_TAGS,
            autoscroll=AUTOSCROLL,
            buttons=CHATBOT_BUTTONS,
            placeholder=CHAT_PLACEHOLDER,
        )
        with gr.Row():
            textbox = gr.Textbox(
                placeholder=INPUT_PLACEHOLDER,
                show_label=False,
                lines=INPUT_LINES,
                max_lines=INPUT_MAX_LINES,
                scale=10,
            )
            send_btn = gr.Button(SEND_BUTTON_TEXT, variant="primary", scale=1)
            stop_btn = gr.Button(
                STOP_BUTTON_TEXT, variant="stop", scale=1, visible=False
            )
        clear_btn = gr.Button(CLEAR_BUTTON_TEXT, size="sm")

        outputs = [chatbot, state, textbox, send_btn, stop_btn]
        submitted = gr.on(
            triggers=[send_btn.click, textbox.submit],
            fn=submit,
            inputs=[textbox, state],
            outputs=outputs + [job],
            queue=False,
            api_visibility="private",
        )
        generation = submitted.success(
            fn=respond,
            inputs=[job, state],
            outputs=outputs,
            concurrency_limit=CONCURRENCY_LIMIT,
            api_visibility="private",
        )
        stop_btn.click(
            fn=stop,
            inputs=[state],
            outputs=outputs,
            cancels=[generation],
            queue=False,
            api_visibility="private",
        )
        clear_btn.click(
            fn=clear,
            inputs=[state],
            outputs=outputs,
            cancels=[generation],
            queue=False,
            api_visibility="private",
        )
        # Keep the Chatbot toolbar's built-in clear action in sync with model history.
        chatbot.clear(
            fn=clear,
            inputs=[state],
            outputs=outputs,
            cancels=[generation],
            queue=False,
            api_visibility="private",
        )
    return demo

demo = build_app()
demo.queue(
    default_concurrency_limit=CONCURRENCY_LIMIT,
    max_size=QUEUE_MAX_SIZE,
)

if __name__ == "__main__":
    logging.basicConfig(level=logging.WARNING)
    demo.launch(
        server_name=SERVER_NAME,
        server_port=SERVER_PORT,
        share=SHARE,
        theme=gr.themes.Soft(
            primary_hue="sky",
            secondary_hue="blue",
            neutral_hue="slate",
            radius_size="lg",
            font=[
                "system-ui",
                "-apple-system",
                "BlinkMacSystemFont",
                "Segoe UI",
                "Roboto",
                "Arial",
                "sans-serif",
            ],
            font_mono=[
                "ui-monospace",
                "SFMono-Regular",
                "Menlo",
                "Monaco",
                "Consolas",
                "monospace",
            ],
        ),
        ssr_mode=False,
    )