| """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_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_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") |
|
|
| |
| 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) |
|
|
| |
| 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.", |
| ) |
|
|
| |
| 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." |
| ) |
|
|
| |
| 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.", |
| ) |
|
|
| |
| 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: |
| |
| 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) |
|
|
| |
| 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): |
| |
| 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: |
| |
| 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: |
| |
| 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: |
| |
| raise |
| except Exception as exc: |
| |
| 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 |
| |
| 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: |
| |
| 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", |
| ) |
| |
| 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, |
| ) |
|
|
