Upload 5 files

README.md

@@ -1,96 +1,137 @@
----
-title: React
-emoji: 🌍
-colorFrom: yellow
-colorTo: gray
-sdk: gradio
-sdk_version: 6.9.0
-app_file: app.py
-pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
-
-
-# Nanbeige4.1-3B Inference Server
-
-Lightweight remote LLM inference service for Enterprise ReAct Agent systems.
-
-## Overview
-
-This Hugging Face Space hosts the **Nanbeige4.1-3B** model as a remote inference API, designed to work with local agent orchestration systems. The model runs entirely in this Space, while all agent logic, tools, and memory systems run on the user's local machine.
-
-## Model Information
-
-- **Model**: [Nanbeige/Nanbeige4.1-3B](https://huggingface.co/Nanbeige/Nanbeige4.1-3B)
-- **Parameters**: 3B
-- **Context Window**: 8K tokens
-- **Capabilities**: Tool calling, reasoning, 500+ tool invocation rounds
-- **License**: Apache 2.0
-
-## API Endpoints
-
-### POST /chat
-Main chat completion endpoint (OpenAI-compatible).
-
-**Request:**
-```json
-{
-  "messages": [
-    {"role": "system", "content": "You are a helpful assistant."},
-    {"role": "user", "content": "Hello!"}
-  ],
-  "tools": [...],
-  "stream": false,
-  "max_tokens": 2048,
-  "temperature": 0.6,
-  "top_p": 0.95
-}
-```
-
-**Response:**
-```json
-{
-  "id": "chatcmpl-...",
-  "object": "chat.completion",
-  "created": 1234567890,
-  "model": "Nanbeige/Nanbeige4.1-3B",
-  "choices": [...],
-  "usage": {
-    "prompt_tokens": 20,
-    "completion_tokens": 50,
-    "total_tokens": 70
-  }
-}
-```
-
-### GET /chat
-Web interface for testing.
-
-### GET /health
-Health check endpoint.
-
-## Usage with Local Agent
-
-```python
-import requests
-
-response = requests.post(
-    "https://your-space.hf.space/chat",
-    json={
-        "messages": [{"role": "user", "content": "Hello!"}],
-        "temperature": 0.6
-    }
-)
-result = response.json()
-```
-
-## Hardware Requirements
-
-- **GPU**: Recommended (CUDA-compatible)
-- **CPU**: Fallback supported
-- **Memory**: ~8GB RAM minimum
-
-## Local Agent Repository
-
-For the complete local agent system that connects to this Space, see the companion repository.
+# HF Space Backend (Streaming LLM Server)
+
+This folder contains Hugging Face Space backends for two model deployments that share the same production runtime.
+
+## Files
+- `app.py`: Nanbeige deployment entrypoint (`Nanbeige/Nanbeige4.1-3B`)
+- `main.py`: LiquidAI deployment entrypoint (`LiquidAI/LFM2.5-1.2B-Thinking`)
+- `server_runtime.py`: shared queue + worker + streaming runtime used by both entrypoints
+- `index.html`: lightweight local streaming test UI
+- `requirements.txt`: runtime dependencies
+
+## Runtime Architecture
+Both servers use the same execution flow:
+
+Client Request  
+-> FastAPI `/chat`  
+-> `asyncio.Queue` request buffer  
+-> worker pool (`asyncio` tasks)  
+-> concurrency gate (`asyncio.Semaphore`)  
+-> one generation thread per request (`model.generate`)  
+-> per-request `TextIteratorStreamer`  
+-> SSE token stream to client
+
+### Why this structure
+- Keeps the event loop responsive.
+- Prevents response mixing across users (isolated request objects).
+- Supports controlled concurrency under CPU/GPU.
+- Queues overflow load instead of hard failing during bursts.
+
+## Concurrency
+Hardware-aware worker count:
+- CPU: `1..4` workers (core-based)
+- GPU: `3..5` workers (VRAM tier-based)
+
+Override at runtime:
+- `HF_MAX_WORKERS`
+
+Queue settings:
+- `HF_QUEUE_MAX_SIZE` (default: `512`)
+
+## Thread Lifecycle and Safety
+- Each request gets its own generation thread.
+- Each request has a cancellation event.
+- `CancelAwareStoppingCriteria` stops generation when client disconnects/cancels.
+- Streamer is explicitly ended in `finally` block.
+- Generation thread is joined with a long timeout (`HF_GENERATION_JOIN_TIMEOUT_SECONDS`, default `180`) to avoid orphaned work.
+
+This fixes the old short-join behavior that produced frequent:
+- `Generation thread did not finish within timeout`
+
+## Metrics and Logging
+Per request logs include:
+- request queued
+- worker start/end
+- first token latency
+- generated token count
+- tokens/sec
+- active workers
+- queue size
+
+Debug token-by-token logging is optional:
+- `HF_DEBUG_TOKEN_LOGS=1`
+
+## API
+### `POST /chat`
+Body:
+- `messages`: chat messages
+- `stream`: `true` for SSE streaming
+- `max_tokens`: max new tokens requested
+- `temperature`: optional; if omitted model default is used
+- `tools`: optional tool schemas for chat template
+
+Streaming response format:
+- SSE `data: {"type":"token","content":"..."}` chunks
+- final `{"type":"done","content":""}` event
+
+### `GET /health`
+Returns:
+- `status`
+- `model_loaded`
+- `device`
+- `active_workers`
+- `queue_size`
+- `max_workers`
+
+### `GET /index`
+Serves `index.html` test page.
+
+## Model-Specific Settings
+### `app.py` (Nanbeige4.1-3B)
+- `max_input_tokens=32768`
+- `eos_token_id=166101`
+- `default_temperature=0.6`
+- `top_p=0.95`
+- `repetition_penalty=1.0`
+- `tokenizer_use_fast=False`
+
+### `main.py` (LFM2.5-1.2B-Thinking)
+- `max_input_tokens=32768`
+- `default_temperature=0.1`
+- `top_p=0.1`
+- `top_k=50`
+- `repetition_penalty=1.05`
+- `eos_token_id` from tokenizer config
+
+## Environment Variables
+- `HF_MAX_WORKERS`
+- `HF_QUEUE_MAX_SIZE`
+- `HF_STREAMER_TIMEOUT_SECONDS`
+- `HF_GENERATION_JOIN_TIMEOUT_SECONDS`
+- `HF_MAX_INPUT_TOKENS`
+- `HF_MAX_NEW_TOKENS`
+- `HF_DEBUG_TOKEN_LOGS`
+
+## Model Documentation References
+### Nanbeige / `app.py`
+- https://huggingface.co/Nanbeige/Nanbeige4.1-3B
+- https://huggingface.co/Nanbeige/Nanbeige4.1-3B/blob/main/README.md
+- https://huggingface.co/Nanbeige/Nanbeige4.1-3B/blob/main/Nanbeige4.1-3B-Report.pdf
+- https://huggingface.co/Nanbeige/Nanbeige4.1-3B/blob/main/generation_config.json
+- https://huggingface.co/Nanbeige/Nanbeige4.1-3B/blob/main/config.json
+
+### LiquidAI / `main.py`
+- https://huggingface.co/LiquidAI/LFM2.5-1.2B-Thinking
+- https://huggingface.co/LiquidAI/LFM2.5-1.2B-Thinking/blob/main/README.md
+- https://huggingface.co/LiquidAI/LFM2.5-1.2B-Thinking/blob/main/chat_template.jinja
+- https://huggingface.co/LiquidAI/LFM2.5-1.2B-Thinking/blob/main/config.json
+- https://docs.liquid.ai/lfm/key-concepts/chat-template
+- https://docs.liquid.ai/lfm/key-concepts/text-generation-and-prompting
+- https://docs.liquid.ai/lfm/key-concepts/tool-use
+- https://huggingface.co/docs/transformers/en/chat_templating#using-applychattemplate
+
+## Notes
+- Model is loaded once per process during FastAPI lifespan startup.
+- `index.html` is intentionally a simple streaming test page, not the production frontend.
+- Both entrypoints (`app.py`, `main.py`) now behave consistently by design.
+

app.py

@@ -1,294 +1,37 @@
 """
-HuggingFace Space application for Nanbeige4.1-3B model inference.
-Provides streaming chat completion API.
-"""
+Hugging Face Space server for Nanbeige/Nanbeige4.1-3B.
 
-import os
-import json
-import asyncio
-import time
-from typing import AsyncGenerator, List, Dict, Any, Optional
-from contextlib import asynccontextmanager
-from datetime import datetime
+This file uses the shared runtime with:
+- async queue buffering
+- worker pool + semaphore concurrency
+- safe per-request generation thread lifecycle
+"""
 
-from fastapi import FastAPI, HTTPException
-from fastapi.responses import StreamingResponse
-from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import FileResponse
-from pydantic import BaseModel
-from transformers import AutoTokenizer, AutoModelForCausalLM, TextIteratorStreamer
-from threading import Thread
-import torch
-import logging
+try:
+    from .server_runtime import RuntimeConfig, create_hf_space_app
+except ImportError:  # pragma: no cover - direct script execution
+    from server_runtime import RuntimeConfig, create_hf_space_app
 
-logger = logging.getLogger(__name__)
-logging.basicConfig(level=logging.INFO)
 
-# Model configuration
 MODEL_NAME = "Nanbeige/Nanbeige4.1-3B"
-MAX_LENGTH = 32768
-
-# Global model and tokenizer
-model = None
-tokenizer = None
-
-BASE_DIR = os.path.dirname(os.path.abspath(__file__))
-
-class Message(BaseModel):
-    role: str
-    content: str
-
-
-class ChatRequest(BaseModel):
-    messages: List[Message]
-    stream: bool = True
-    max_tokens: int = 8192  # Increased from 2048 (supports up to 131072)
-    temperature: float = 0.6  # Nanbeige4.1-3B recommended
-    tools: Optional[List[Dict]] = None
-
 
-@asynccontextmanager
-async def lifespan(app: FastAPI):
-    """Application lifespan handler."""
-    global model, tokenizer
-    
-    print("Loading model...")
-    tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME, trust_remote_code=True)
-    model = AutoModelForCausalLM.from_pretrained(
-        MODEL_NAME,
-        trust_remote_code=True,
-        torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
-        device_map="auto" if torch.cuda.is_available() else None
+app = create_hf_space_app(
+    RuntimeConfig(
+        model_name=MODEL_NAME,
+        title="Nanbeige4.1-3B Inference API",
+        description="Streaming chat completion API for Nanbeige4.1-3B",
+        max_input_tokens=32768,
+        eos_token_id=166101,
+        default_temperature=0.6,
+        top_p=0.95,
+        repetition_penalty=1.0,
+        tokenizer_use_fast=False,
+        logger_name=__name__,
     )
-    
-    if not torch.cuda.is_available():
-        model = model.to("cpu")
-    
-    print("Model loaded successfully!")
-    yield
-    
-    # Cleanup
-    del model
-    del tokenizer
-    torch.cuda.empty_cache()
-
-
-app = FastAPI(
-    title="Nanbeige4.1-3B Inference API",
-    description="Streaming chat completion API for Nanbeige4.1-3B",
-    version="1.0.0",
-    lifespan=lifespan
-)
-
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
 )
 
 
-def format_messages_proper(messages: List[Message], tools: Optional[List[Dict]] = None) -> str:
-    """Format messages using the model's proper chat template.
-    
-    Nanbeige4.1-3B uses the HF transformers chat template.
-    This ensures proper formatting for both regular and tool-aware conversations.
-    """
-    global tokenizer
-    
-    # Convert Message objects to dicts for tokenizer
-    message_dicts = [{"role": msg.role, "content": msg.content} for msg in messages]
-    
-    # Use tokenizer's built-in chat template for proper formatting
-    if tools:
-        # Tool-aware formatting (for function calling)
-        prompt = tokenizer.apply_chat_template(
-            message_dicts,
-            tools=tools,
-            add_generation_prompt=True,
-            tokenize=False
-        )
-    else:
-        # Regular chat formatting
-        prompt = tokenizer.apply_chat_template(
-            message_dicts,
-            add_generation_prompt=True,
-            tokenize=False
-        )
-    
-    return prompt
-
-
-async def stream_tokens(prompt: str, max_tokens: int, temperature: float, tools: Optional[List[Dict]] = None) -> AsyncGenerator[str, None]:
-    """Stream tokens from the model token-by-token as fast as generated.
-    
-    Uses Nanbeige4.1-3B recommended hyperparameters.
-    """
-    global model, tokenizer
-    
-    start_time = time.time()
-    logger.info(f"Starting token generation for prompt length: {len(prompt)}")
-    
-    inputs = tokenizer(
-        prompt,
-        return_tensors="pt",
-        truncation=True,
-        max_length=2048
-    )
-    
-    if torch.cuda.is_available():
-        inputs = inputs.to("cuda")
-    
-    # Create streamer with timeout to prevent hanging
-    streamer = TextIteratorStreamer(
-        tokenizer,
-        skip_prompt=True,
-        skip_special_tokens=True,
-        timeout=300.0  # 5 min timeout per token
-    )
-    
-    generation_kwargs = dict(
-        **inputs,
-        streamer=streamer,
-        max_new_tokens=min(max_tokens, 131072),  # Support up to model's max (131072)
-        temperature=temperature,
-        top_p=0.95,  # Nanbeige4.1-3B recommended
-        repetition_penalty=1.0,  # Nanbeige4.1-3B recommended
-        do_sample=temperature > 0,
-        eos_token_id=166101,  # Nanbeige4.1-3B specific EOS token
-        pad_token_id=tokenizer.eos_token_id
-    )
-    
-    # Run generation in separate thread (non-blocking)
-    thread = Thread(target=model.generate, kwargs=generation_kwargs, daemon=False)
-    thread.start()
-    
-    generated_text = ""
-    token_count = 0
-    first_token_time = None
-    
-    try:
-        for new_text in streamer:
-            if new_text:  # Skip empty strings
-                generated_text += new_text
-                token_count += 1
-                
-                # Log first token time (time to first byte)
-                if first_token_time is None:
-                    first_token_time = time.time() - start_time
-                    logger.info(f"First token generated in {first_token_time:.2f}s")
-                
-                # preview logging to verify streaming works
-                logger.info(f"streaming token #{token_count}: {repr(new_text)}")
-
-                # Yield SSE event immediately (no buffering)
-                data = json.dumps({"type": "token", "content": new_text})
-                yield f"data: {data}\n\n"
-                # let the event loop schedule a send/flush so proxies don't buffer
-                await asyncio.sleep(0)
-                logger.debug(f"Token {token_count}: {repr(new_text[:20])}...")
-        
-        # Log generation stats
-        total_time = time.time() - start_time
-        tokens_per_sec = token_count / total_time if total_time > 0 else 0
-        logger.info(f"Generation complete: {token_count} tokens in {total_time:.2f}s ({tokens_per_sec:.2f} tok/s)")
-        
-        # Signal completion
-        yield f"data: {json.dumps({'type': 'done', 'content': ''})}\n\n"
-    
-    except Exception as e:
-        logger.error(f"Token generation error: {e}", exc_info=True)
-        yield f"data: {json.dumps({'type': 'error', 'content': str(e)})}\n\n"
-    
-    finally:
-        # Wait for thread to finish
-        thread.join(timeout=5)
-        if thread.is_alive():
-            logger.warning("Generation thread did not finish within timeout")
-
-
-@app.get("/")
-async def root():
-    """Root endpoint."""
-    return {
-        "name": "Nanbeige4.1-3B Inference API",
-        "version": "1.0.0",
-        "model": MODEL_NAME,
-        "status": "running"
-    }
-
-
-@app.get("/index", response_class=FileResponse)
-async def serve_chat():
-    """Serve chat.html as index."""
-    return FileResponse(os.path.join(BASE_DIR, "index.html"))
-
-
-@app.get("/health")
-async def health():
-    """Health check endpoint."""
-    return {
-        "status": "healthy",
-        "model_loaded": model is not None and tokenizer is not None
-    }
-
-
-@app.post("/chat")
-async def chat(request: ChatRequest):
-    """
-    Chat completion endpoint with streaming support.
-    """
-    if model is None or tokenizer is None:
-        raise HTTPException(status_code=503, detail="Model not loaded yet")
-    
-    # Format messages using the model's proper chat template
-    prompt = format_messages_proper(request.messages, request.tools)
-    
-    if request.stream:
-        # Return streaming response with anti-buffering headers
-        return StreamingResponse(
-            stream_tokens(prompt, request.max_tokens, request.temperature, request.tools),
-            media_type="text/event-stream",
-            headers={
-                "Cache-Control": "no-cache, no-store, must-revalidate",
-                "Pragma": "no-cache",
-                "Expires": "0",
-                "Connection": "keep-alive",
-                "X-Accel-Buffering": "no",
-                "Transfer-Encoding": "chunked"
-            }
-        )
-    else:
-        # Non-streaming response
-        inputs = tokenizer(prompt, return_tensors="pt")
-        if torch.cuda.is_available():
-            inputs = inputs.to("cuda")
-        
-        outputs = model.generate(
-            **inputs,
-            max_new_tokens=min(request.max_tokens, 131072),  # Support up to model's max
-            temperature=request.temperature,
-            top_p=0.95,  # Nanbeige4.1-3B recommended
-            repetition_penalty=1.0,  # Nanbeige4.1-3B recommended
-            do_sample=request.temperature > 0,
-            eos_token_id=166101,  # Model-specific EOS token
-            pad_token_id=tokenizer.eos_token_id
-        )
-        
-        response_text = tokenizer.decode(outputs[0], skip_special_tokens=True)
-        # Extract only the assistant's response
-        response_text = response_text[len(prompt):].strip()
-        
-        return {
-            "content": response_text,
-            "usage": {
-                "prompt_tokens": inputs.input_ids.shape[1],
-                "completion_tokens": outputs.shape[1] - inputs.input_ids.shape[1]
-            }
-        }
-
-
 if __name__ == "__main__":
     import uvicorn
-    uvicorn.run(app, host="0.0.0.0", port=7860)
+
+    uvicorn.run(app, host="0.0.0.0", port=7860)

server_runtime.py

@@ -0,0 +1,522 @@
+"""
+Shared Hugging Face Space runtime for streaming chat inference.
+
+This module provides:
+- one-time global model loading
+- async request queue
+- worker pool with semaphore-based concurrency limits
+- per-request streamer/thread isolation
+- SSE streaming responses
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import time
+import uuid
+from contextlib import asynccontextmanager
+from dataclasses import dataclass, field
+from queue import Empty as QueueEmpty
+from threading import Event as ThreadEvent
+from threading import Thread
+from typing import Any, Dict, List, Optional
+
+import torch
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import FileResponse, StreamingResponse
+from pydantic import BaseModel
+from transformers import (
+    AutoModelForCausalLM,
+    AutoTokenizer,
+    StoppingCriteria,
+    StoppingCriteriaList,
+    TextIteratorStreamer,
+)
+
+
+class Message(BaseModel):
+    role: str
+    content: str
+
+
+class ChatRequest(BaseModel):
+    messages: List[Message]
+    stream: bool = True
+    max_tokens: int = 8192
+    temperature: Optional[float] = None
+    tools: Optional[List[Dict[str, Any]]] = None
+
+
+@dataclass(frozen=True)
+class RuntimeConfig:
+    model_name: str
+    title: str
+    description: str
+    version: str = "1.0.0"
+    max_input_tokens: int = 32768
+    max_new_tokens: int = 131072
+    top_p: float = 0.95
+    top_k: Optional[int] = None
+    repetition_penalty: float = 1.0
+    eos_token_id: Optional[int] = None
+    default_temperature: float = 0.6
+    tokenizer_use_fast: Optional[bool] = None
+    logger_name: str = "hf_space"
+
+
+@dataclass
+class GenerationTask:
+    request_id: str
+    prompt: str
+    max_tokens: int
+    temperature: float
+    output_queue: asyncio.Queue[Optional[Dict[str, Any]]]
+    created_at: float = field(default_factory=time.time)
+    cancel_event: ThreadEvent = field(default_factory=ThreadEvent)
+    prompt_tokens: int = 0
+    generated_tokens: int = 0
+    first_token_latency: Optional[float] = None
+    start_time: Optional[float] = None
+    end_time: Optional[float] = None
+
+
+class CancelAwareStoppingCriteria(StoppingCriteria):
+    """Stops generation when the request is cancelled/disconnected."""
+
+    def __init__(self, cancel_event: ThreadEvent):
+        self.cancel_event = cancel_event
+
+    def __call__(self, input_ids, scores, **kwargs) -> bool:
+        return self.cancel_event.is_set()
+
+
+def _is_truthy(value: str) -> bool:
+    return value.strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _format_sse_event(payload: Dict[str, Any]) -> str:
+    return f"data: {json.dumps(payload)}\n\n"
+
+
+def _detect_concurrency(device: str) -> int:
+    # Allow environment override if needed for debugging/tuning.
+    override = os.getenv("HF_MAX_WORKERS", "").strip()
+    if override:
+        try:
+            parsed = int(override)
+            if parsed > 0:
+                return parsed
+        except ValueError:
+            pass
+
+    if device == "cuda" and torch.cuda.is_available():
+        total_vram_gb = torch.cuda.get_device_properties(0).total_memory / (1024 ** 3)
+        if total_vram_gb >= 20:
+            return 5
+        if total_vram_gb >= 10:
+            return 4
+        return 3
+
+    cpu_count = os.cpu_count() or 1
+    return max(1, min(4, max(1, cpu_count // 2)))
+
+
+def create_hf_space_app(config: RuntimeConfig) -> FastAPI:
+    logger = logging.getLogger(config.logger_name)
+    logging.basicConfig(level=logging.INFO)
+
+    debug_token_logs = _is_truthy(os.getenv("HF_DEBUG_TOKEN_LOGS", "0"))
+    queue_max_size = int(os.getenv("HF_QUEUE_MAX_SIZE", "512"))
+    streamer_timeout = float(os.getenv("HF_STREAMER_TIMEOUT_SECONDS", "8"))
+    join_timeout = float(os.getenv("HF_GENERATION_JOIN_TIMEOUT_SECONDS", "180"))
+    max_input_tokens = int(os.getenv("HF_MAX_INPUT_TOKENS", str(config.max_input_tokens)))
+    max_new_tokens_limit = int(os.getenv("HF_MAX_NEW_TOKENS", str(config.max_new_tokens)))
+
+    base_dir = os.path.dirname(os.path.abspath(__file__))
+
+    model = None
+    tokenizer = None
+    device = "cuda" if torch.cuda.is_available() else "cpu"
+    max_workers = _detect_concurrency(device)
+
+    request_queue: asyncio.Queue[Optional[GenerationTask]] = asyncio.Queue(maxsize=queue_max_size)
+    worker_tasks: List[asyncio.Task] = []
+    worker_semaphore = asyncio.Semaphore(max_workers)
+
+    active_workers = 0
+    active_workers_lock = asyncio.Lock()
+
+    async def set_active_workers(delta: int) -> int:
+        nonlocal active_workers
+        async with active_workers_lock:
+            active_workers += delta
+            if active_workers < 0:
+                active_workers = 0
+            return active_workers
+
+    def format_messages_proper(messages: List[Message], tools: Optional[List[Dict[str, Any]]] = None) -> str:
+        message_dicts = [{"role": msg.role, "content": msg.content} for msg in messages]
+        if tools:
+            return tokenizer.apply_chat_template(
+                message_dicts,
+                tools=tools,
+                add_generation_prompt=True,
+                tokenize=False,
+            )
+        return tokenizer.apply_chat_template(
+            message_dicts,
+            add_generation_prompt=True,
+            tokenize=False,
+        )
+
+    async def run_generation(task: GenerationTask, worker_id: int) -> None:
+        request_start = time.time()
+        task.start_time = request_start
+        await set_active_workers(+1)
+
+        try:
+            logger.info(
+                "[%s] worker=%d start queue_size=%d active_workers=%d",
+                task.request_id,
+                worker_id,
+                request_queue.qsize(),
+                active_workers,
+            )
+
+            inputs = tokenizer(
+                task.prompt,
+                return_tensors="pt",
+                truncation=True,
+                max_length=max_input_tokens,
+                add_special_tokens=False,
+            )
+
+            task.prompt_tokens = int(inputs.input_ids.shape[1])
+
+            if device == "cuda":
+                inputs = inputs.to("cuda")
+
+            streamer = TextIteratorStreamer(
+                tokenizer,
+                skip_prompt=True,
+                skip_special_tokens=True,
+                timeout=streamer_timeout,
+            )
+
+            stopping_criteria = StoppingCriteriaList(
+                [CancelAwareStoppingCriteria(task.cancel_event)]
+            )
+
+            generation_kwargs: Dict[str, Any] = dict(
+                **inputs,
+                streamer=streamer,
+                max_new_tokens=min(task.max_tokens, max_new_tokens_limit),
+                temperature=task.temperature,
+                top_p=config.top_p,
+                repetition_penalty=config.repetition_penalty,
+                do_sample=task.temperature > 0,
+                eos_token_id=config.eos_token_id if config.eos_token_id is not None else tokenizer.eos_token_id,
+                pad_token_id=tokenizer.eos_token_id,
+                stopping_criteria=stopping_criteria,
+            )
+            if config.top_k is not None:
+                generation_kwargs["top_k"] = config.top_k
+
+            generation_error: Dict[str, Exception] = {}
+            generation_done = ThreadEvent()
+
+            def generate_target() -> None:
+                try:
+                    with torch.inference_mode():
+                        model.generate(**generation_kwargs)
+                except Exception as exc:  # pragma: no cover - defensive logging
+                    generation_error["error"] = exc
+                    logger.error("[%s] generation thread error: %s", task.request_id, exc, exc_info=True)
+                finally:
+                    generation_done.set()
+                    try:
+                        streamer.end()
+                    except Exception:
+                        # Best-effort close of streamer queue.
+                        pass
+
+            generation_thread = Thread(
+                target=generate_target,
+                name=f"gen-{task.request_id[:8]}",
+                daemon=True,
+            )
+            generation_thread.start()
+
+            stream_iter = iter(streamer)
+            while True:
+                if task.cancel_event.is_set():
+                    logger.info("[%s] cancellation requested", task.request_id)
+                    break
+
+                try:
+                    new_text = await asyncio.to_thread(next, stream_iter)
+                except StopIteration:
+                    break
+                except QueueEmpty:
+                    if generation_done.is_set():
+                        break
+                    continue
+                except Exception as exc:  # pragma: no cover - defensive logging
+                    if generation_done.is_set():
+                        break
+                    logger.error("[%s] streamer read error: %s", task.request_id, exc, exc_info=True)
+                    generation_error["error"] = exc
+                    break
+
+                if not new_text:
+                    continue
+
+                task.generated_tokens += 1
+                if task.first_token_latency is None:
+                    task.first_token_latency = time.time() - request_start
+                    logger.info(
+                        "[%s] first_token=%.2fs worker=%d",
+                        task.request_id,
+                        task.first_token_latency,
+                        worker_id,
+                    )
+
+                if debug_token_logs:
+                    logger.info("[%s] token#%d: %r", task.request_id, task.generated_tokens, new_text)
+
+                await task.output_queue.put({"type": "token", "content": new_text})
+                await asyncio.sleep(0)
+
+            # Ensure generation thread is not left running in background.
+            try:
+                await asyncio.wait_for(asyncio.to_thread(generation_thread.join), timeout=join_timeout)
+            except asyncio.TimeoutError:
+                logger.error(
+                    "[%s] generation thread still alive after %.1fs join timeout",
+                    task.request_id,
+                    join_timeout,
+                )
+
+            if task.cancel_event.is_set():
+                await task.output_queue.put({"type": "error", "content": "Generation interrupted. You can continue."})
+            elif "error" in generation_error:
+                await task.output_queue.put({"type": "error", "content": str(generation_error["error"])})
+            else:
+                await task.output_queue.put({"type": "done", "content": ""})
+
+        except Exception as exc:
+            logger.error("[%s] worker failure: %s", task.request_id, exc, exc_info=True)
+            await task.output_queue.put({"type": "error", "content": str(exc)})
+        finally:
+            task.end_time = time.time()
+            duration = max(1e-6, task.end_time - request_start)
+            tps = task.generated_tokens / duration
+            logger.info(
+                "[%s] worker=%d end tokens=%d duration=%.2fs tok_s=%.2f active_workers=%d queue_size=%d",
+                task.request_id,
+                worker_id,
+                task.generated_tokens,
+                duration,
+                tps,
+                active_workers,
+                request_queue.qsize(),
+            )
+
+            await task.output_queue.put(None)
+            await set_active_workers(-1)
+
+    async def worker_loop(worker_id: int) -> None:
+        logger.info("Worker-%d started", worker_id)
+        while True:
+            task = await request_queue.get()
+            if task is None:
+                request_queue.task_done()
+                logger.info("Worker-%d received shutdown signal", worker_id)
+                break
+
+            try:
+                if task.cancel_event.is_set():
+                    await task.output_queue.put({"type": "error", "content": "Request cancelled before execution."})
+                    await task.output_queue.put(None)
+                    continue
+
+                async with worker_semaphore:
+                    await run_generation(task, worker_id)
+            finally:
+                request_queue.task_done()
+
+        logger.info("Worker-%d stopped", worker_id)
+
+    @asynccontextmanager
+    async def lifespan(app: FastAPI):
+        nonlocal model, tokenizer, worker_tasks, max_workers, device
+
+        logger.info("Loading model %s on %s", config.model_name, device)
+        tokenizer_kwargs: Dict[str, Any] = {"trust_remote_code": True}
+        if config.tokenizer_use_fast is not None:
+            tokenizer_kwargs["use_fast"] = config.tokenizer_use_fast
+        tokenizer = AutoTokenizer.from_pretrained(config.model_name, **tokenizer_kwargs)
+        model = AutoModelForCausalLM.from_pretrained(
+            config.model_name,
+            trust_remote_code=True,
+            torch_dtype="auto" if device == "cuda" else torch.float32,
+            device_map="auto" if device == "cuda" else None,
+        )
+
+        if device != "cuda":
+            model = model.to("cpu")
+
+        logger.info(
+            "Model loaded: %s | device=%s | max_workers=%d | queue_max_size=%d",
+            config.model_name,
+            device,
+            max_workers,
+            queue_max_size,
+        )
+        logger.info(
+            "Runtime config: max_input_tokens=%d max_new_tokens_limit=%d top_p=%.3f top_k=%s rep_penalty=%.3f",
+            max_input_tokens,
+            max_new_tokens_limit,
+            config.top_p,
+            str(config.top_k),
+            config.repetition_penalty,
+        )
+
+        worker_tasks = [
+            asyncio.create_task(worker_loop(i + 1), name=f"generation-worker-{i + 1}")
+            for i in range(max_workers)
+        ]
+
+        try:
+            yield
+        finally:
+            logger.info("Shutting down workers...")
+            for _ in worker_tasks:
+                await request_queue.put(None)
+            await asyncio.gather(*worker_tasks, return_exceptions=True)
+
+            logger.info("Releasing model resources...")
+            del model
+            del tokenizer
+            if torch.cuda.is_available():
+                torch.cuda.empty_cache()
+
+    app = FastAPI(
+        title=config.title,
+        description=config.description,
+        version=config.version,
+        lifespan=lifespan,
+    )
+
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["*"],
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    @app.get("/")
+    async def root():
+        return {
+            "name": config.title,
+            "version": config.version,
+            "model": config.model_name,
+            "status": "running",
+            "device": device,
+            "max_workers": max_workers,
+        }
+
+    @app.get("/index", response_class=FileResponse)
+    async def serve_chat():
+        return FileResponse(os.path.join(base_dir, "index.html"))
+
+    @app.get("/health")
+    async def health():
+        return {
+            "status": "healthy",
+            "model_loaded": model is not None and tokenizer is not None,
+            "device": device,
+            "active_workers": active_workers,
+            "queue_size": request_queue.qsize(),
+            "max_workers": max_workers,
+        }
+
+    @app.post("/chat")
+    async def chat(request: ChatRequest):
+        if model is None or tokenizer is None:
+            raise HTTPException(status_code=503, detail="Model not loaded yet")
+
+        prompt = format_messages_proper(request.messages, request.tools)
+        task = GenerationTask(
+            request_id=uuid.uuid4().hex,
+            prompt=prompt,
+            max_tokens=request.max_tokens,
+            temperature=request.temperature if request.temperature is not None else config.default_temperature,
+            output_queue=asyncio.Queue(maxsize=2048),
+        )
+
+        logger.info(
+            "[%s] queued request prompt_len=%d queue_size=%d",
+            task.request_id,
+            len(prompt),
+            request_queue.qsize(),
+        )
+        await request_queue.put(task)
+
+        if request.stream:
+            async def stream_events():
+                try:
+                    while True:
+                        event = await task.output_queue.get()
+                        if event is None:
+                            break
+                        yield _format_sse_event(event)
+                except asyncio.CancelledError:
+                    task.cancel_event.set()
+                    raise
+                finally:
+                    task.cancel_event.set()
+
+            return StreamingResponse(
+                stream_events(),
+                media_type="text/event-stream",
+                headers={
+                    "Cache-Control": "no-cache, no-store, must-revalidate",
+                    "Pragma": "no-cache",
+                    "Expires": "0",
+                    "Connection": "keep-alive",
+                    "X-Accel-Buffering": "no",
+                    "Transfer-Encoding": "chunked",
+                },
+            )
+
+        chunks: List[str] = []
+        error_message: Optional[str] = None
+        while True:
+            event = await task.output_queue.get()
+            if event is None:
+                break
+            event_type = event.get("type")
+            if event_type == "token":
+                chunks.append(str(event.get("content", "")))
+            elif event_type == "error":
+                error_message = str(event.get("content", "Generation failed"))
+
+        if error_message:
+            raise HTTPException(status_code=500, detail=error_message)
+
+        response_text = "".join(chunks).strip()
+        return {
+            "content": response_text,
+            "usage": {
+                "prompt_tokens": task.prompt_tokens,
+                "completion_tokens": task.generated_tokens,
+            },
+        }
+
+    return app