from fastapi import FastAPI, HTTPException from fastapi.responses import JSONResponse from transformers import AutoTokenizer, AutoModelForCausalLM import torch import time import asyncio import json import re from typing import Dict, Any, Optional import logging import traceback # Setup logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI(title="Qwen3 API", description="OpenAI-compatible API for Qwen3 models", version="1.0.0") # Global variables models = {} tokenizers = {} MODEL_CONFIGS = { "qwen3-1.7b": "Qwen/Qwen3-1.7B", "qwen3-4b": "Qwen/Qwen3-4B" } def download_model_safely(model_name: str, max_retries: int = 3): """Download model với retry logic""" for attempt in range(max_retries): try: logger.info(f"Downloading {model_name} (attempt {attempt + 1}/{max_retries})...") tokenizer = AutoTokenizer.from_pretrained( model_name, trust_remote_code=True ) model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype=torch.float16, device_map="auto", trust_remote_code=True, low_cpu_mem_usage=True ) logger.info(f"Successfully loaded {model_name}") return tokenizer, model except Exception as e: logger.error(f"Download failed (attempt {attempt + 1}): {str(e)}") if attempt == max_retries - 1: raise e time.sleep(30) def load_model_on_demand(model_key: str): """Load model khi cần thiết""" if model_key not in models: if model_key not in MODEL_CONFIGS: raise ValueError(f"Unknown model key: {model_key}") model_name = MODEL_CONFIGS[model_key] logger.info(f"Loading {model_name} on demand...") # Clear memory if len(models) >= 1: for key in list(models.keys()): logger.info(f"Unloading {key} to free memory...") del models[key] del tokenizers[key] if torch.cuda.is_available(): torch.cuda.empty_cache() import gc gc.collect() tokenizer, model = download_model_safely(model_name) tokenizers[model_key] = tokenizer models[model_key] = model logger.info(f"{model_name} loaded successfully!") def extract_json_from_response(text: str) -> str: """Extract JSON from response text""" # Remove thinking tags completely text = re.sub(r'.*?', '', text, flags=re.DOTALL) text = text.strip() # Try to find JSON object json_match = re.search(r'\{[^{}]*\}', text) if json_match: return json_match.group(0) # If no JSON found, return the cleaned text return text def format_structured_prompt(messages: list, json_schema: dict) -> str: """Format messages with JSON schema instructions""" # Extract schema properties for clear instructions schema_info = json_schema.get('schema', {}) properties = schema_info.get('properties', {}) required = schema_info.get('required', []) # Create clear JSON format instructions json_instructions = f""" You must respond with a valid JSON object only. No explanations, no markdown, no additional text. Required JSON format: {json.dumps(schema_info, indent=2)} Example response format: {{"type": "examschedule"}} """ # Build the conversation formatted_messages = [] for msg in messages: if msg["role"] == "system": # Append JSON instructions to system message content = msg["content"] + "\n" + json_instructions formatted_messages.append({"role": "system", "content": content}) else: formatted_messages.append(msg) return formatted_messages @app.on_event("startup") async def load_models(): """Load default model""" try: logger.info("Loading default model: Qwen3-1.7B...") tokenizer, model = download_model_safely("Qwen/Qwen3-1.7B") tokenizers["qwen3-1.7b"] = tokenizer models["qwen3-1.7b"] = model logger.info("Default model loaded successfully!") except Exception as e: logger.error(f"Failed to load default model: {str(e)}") logger.info("Server will continue running, models will be loaded on demand") @app.get("/") def health_check(): """Health check endpoint""" return { "status": "API is running", "available_models": list(MODEL_CONFIGS.keys()), "loaded_models": list(models.keys()), "version": "1.0.0", "message": "Qwen3 API Service - OpenAI Compatible with Structured Output" } @app.get("/models") def list_models(): """List available models""" return { "available_models": MODEL_CONFIGS, "loaded_models": list(models.keys()), "total_available": len(MODEL_CONFIGS), "total_loaded": len(models) } @app.post("/v1/chat/completions") async def chat_completions(request: Dict[str, Any]): """OpenAI-compatible chat completions endpoint với Structured Output support""" try: logger.info("=== CHAT COMPLETIONS REQUEST START ===") logger.info(f"Request payload: {json.dumps(request, ensure_ascii=False, indent=2)}") # Parse request parameters model_name = request.get("model", "qwen3-1.7b") messages = request.get("messages", []) temperature = request.get("temperature", 0.7) max_tokens = request.get("max_tokens", 200) response_format = request.get("response_format", None) logger.info(f"Model: {model_name}, Temperature: {temperature}, Max tokens: {max_tokens}") logger.info(f"Response format: {response_format}") # Validate input if not messages: logger.error("Messages is empty") raise HTTPException(status_code=400, detail="Messages cannot be empty") # Determine model key if "4b" in model_name.lower() or "4" in model_name.lower(): model_key = "qwen3-4b" else: model_key = "qwen3-1.7b" logger.info(f"Using model key: {model_key}") # Load model if needed if model_key not in models: logger.info(f"Model {model_key} not loaded, loading on demand...") load_model_on_demand(model_key) # Get model and tokenizer tokenizer = tokenizers[model_key] model = models[model_key] logger.info(f"Got tokenizer and model for {model_key}") # Handle structured output if response_format and response_format.get("type") == "json_schema": json_schema = response_format.get("json_schema", {}) logger.info("Structured output requested, formatting messages with JSON schema") messages = format_structured_prompt(messages, json_schema) # Format messages - FORCE DISABLE thinking mode logger.info("Formatting messages with apply_chat_template...") try: text = tokenizer.apply_chat_template( messages, tokenize=False, add_generation_prompt=True, enable_thinking=False # CRITICAL: Force disable thinking ) # AGGRESSIVE thinking mode removal if "" in text or "think>" in text: logger.warning("Found thinking tags in formatted text, removing...") text = re.sub(r'.*?', '', text, flags=re.DOTALL) text = re.sub(r'\s*', '', text) text = text.replace("", "").replace("", "") logger.info(f"Formatted text (first 300 chars): {text[:300]}...") except Exception as e: logger.error(f"Error in apply_chat_template: {str(e)}") # Fallback to simple format WITHOUT thinking text = "" for msg in messages: if msg["role"] == "system": text += f"<|im_start|>system\n{msg['content']}<|im_end|>\n" elif msg["role"] == "user": text += f"<|im_start|>user\n{msg['content']}<|im_end|>\n" elif msg["role"] == "assistant": text += f"<|im_start|>assistant\n{msg['content']}<|im_end|>\n" text += "<|im_start|>assistant\n" # NO thinking tags logger.info(f"Using fallback formatting") # Tokenize input logger.info("Tokenizing input...") model_inputs = tokenizer([text], return_tensors="pt") logger.info(f"Input tokens shape: {model_inputs.input_ids.shape}") # Move to device if hasattr(model, 'device'): logger.info(f"Moving inputs to device: {model.device}") model_inputs = {k: v.to(model.device) for k, v in model_inputs.items()} # Generate response với timeout logger.info("Starting generation...") start_time = time.time() try: # Sử dụng asyncio timeout async def generate_with_timeout(): with torch.no_grad(): generated_ids = model.generate( **model_inputs, max_new_tokens=min(max_tokens, 200), temperature=temperature, do_sample=True if temperature > 0 else False, pad_token_id=tokenizer.eos_token_id, eos_token_id=tokenizer.eos_token_id, repetition_penalty=1.1, top_p=0.9 if temperature > 0 else None, use_cache=True ) return generated_ids # 30 second timeout generated_ids = await asyncio.wait_for(generate_with_timeout(), timeout=30.0) generation_time = time.time() - start_time logger.info(f"Generation completed in {generation_time:.2f} seconds") except asyncio.TimeoutError: logger.error("Generation timeout after 30 seconds") return { "choices": [{ "message": { "content": "Generation timeout. Please try a shorter prompt.", "role": "assistant" }, "finish_reason": "timeout", "index": 0 }], "error": "timeout", "model": model_key } except Exception as e: logger.error(f"Generation error: {str(e)}") logger.error(f"Traceback: {traceback.format_exc()}") return { "choices": [{ "message": { "content": f"Generation error: {str(e)}", "role": "assistant" }, "finish_reason": "error", "index": 0 }], "error": str(e), "model": model_key } # Extract response logger.info("Extracting response...") try: # Get input length correctly if hasattr(model_inputs, 'input_ids'): input_length = model_inputs.input_ids.shape[1] elif isinstance(model_inputs, dict) and 'input_ids' in model_inputs: input_length = model_inputs['input_ids'].shape[1] else: input_length = 0 # Extract output tokens output_ids = generated_ids[0][input_length:].tolist() response = tokenizer.decode(output_ids, skip_special_tokens=True).strip() # Handle structured output if response_format and response_format.get("type") == "json_schema": response = extract_json_from_response(response) logger.info(f"Extracted JSON response: {response}") # Validate JSON try: json.loads(response) except json.JSONDecodeError: logger.warning("Generated response is not valid JSON, attempting to fix...") # Try to extract just the JSON part json_match = re.search(r'\{.*\}', response) if json_match: response = json_match.group(0) else: response = '{"type": "other"}' # Fallback logger.info(f"Final response: {response}") except Exception as e: logger.error(f"Error extracting response: {str(e)}") response = "Error extracting response" # Clean up response if not response: response = "I apologize, but I couldn't generate a proper response. Please try again." # Format response - tương thích với AiService result = { "choices": [{ "message": { "content": response, "role": "assistant" }, "finish_reason": "stop", "index": 0 }], "model": model_key, "usage": { "prompt_tokens": input_length if 'input_length' in locals() else 0, "completion_tokens": len(output_ids) if 'output_ids' in locals() else 0, "total_tokens": (input_length if 'input_length' in locals() else 0) + (len(output_ids) if 'output_ids' in locals() else 0) }, "object": "chat.completion", "created": int(time.time()) } logger.info("=== CHAT COMPLETIONS REQUEST END ===") return result except HTTPException: raise except Exception as e: logger.error(f"Unexpected error in chat_completions: {str(e)}") logger.error(f"Traceback: {traceback.format_exc()}") return { "choices": [{ "message": { "content": f"Unexpected error: {str(e)}", "role": "assistant" }, "finish_reason": "error", "index": 0 }], "error": str(e), "model": "qwen3-1.7b" } @app.get("/health") def health(): """Simple health check""" return { "status": "healthy", "timestamp": int(time.time()), "models_loaded": len(models) } # Error handlers @app.exception_handler(404) async def not_found_handler(request, exc): return JSONResponse( status_code=404, content={ "error": { "message": "Endpoint not found", "type": "not_found_error", "code": 404 } } ) @app.exception_handler(500) async def internal_error_handler(request, exc): return JSONResponse( status_code=500, content={ "error": { "message": "Internal server error", "type": "internal_server_error", "code": 500 } } ) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)