"""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 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, )