""" OPENAI API compatibility utilities """ import time import logging from typing import List, Dict, Any, Optional from pydantic import BaseModel # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Default model configuration DEFAULT_MODEL = "Qwen/Qwen3-Coder-30B-A3B-Instruct" DEFAULT_MAX_TOKENS = 1024 DEFAULT_TEMPERATURE = 0.7 class ChatMessage(BaseModel): """Represents a chat message""" role: str content: str class ChatRequest(BaseModel): """Represents a chat request""" messages: List[ChatMessage] model: str = DEFAULT_MODEL max_tokens: Optional[int] = DEFAULT_MAX_TOKENS temperature: Optional[float] = DEFAULT_TEMPERATURE class ChatResponse(BaseModel): """Represents a chat response""" id: str object: str = "chat.completion" created: int model: str choices: List[Dict[str, Any]] usage: Dict[str, int] def convert_openai_request_to_model_input(request: ChatRequest) -> str: """Convert OPENAI API request to model input format""" prompt = "" for msg in request.messages: if msg.role == "system": prompt += f"System: {msg.content}\n" elif msg.role == "user": prompt += f"User: {msg.content}\n" elif msg.role == "assistant": prompt += f"Assistant: {msg.content}\n" return prompt def create_openai_response(response_text: str, request: ChatRequest) -> ChatResponse: """Create OPENAI API compatible response""" return ChatResponse( id="chatcmpl-" + str(hash(response_text))[:10], created=int(time.time()), model=request.model, choices=[{ "index": 0, "message": { "role": "assistant", "content": response_text }, "finish_reason": "stop" }], usage={ "prompt_tokens": sum(len(msg.content.split()) for msg in request.messages), "completion_tokens": len(response_text.split()), "total_tokens": sum(len(msg.content.split()) for msg in request.messages) + len(response_text.split()) } ) def format_messages_for_frontend(messages: List[Dict[str, Any]]) -> List[Dict[str, str]]: """Format messages for frontend display""" formatted_messages = [] for msg in messages: if msg["role"] == "user": formatted_messages.append({"sender": "user", "text": msg["content"]}) elif msg["role"] == "assistant": formatted_messages.append({"sender": "ai", "text": msg["content"]}) return formatted_messages