utils/api_compat.py

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