|
|
""" |
|
|
OPENAI API compatibility utilities |
|
|
""" |
|
|
|
|
|
import time |
|
|
import logging |
|
|
from typing import List, Dict, Any, Optional |
|
|
from pydantic import BaseModel |
|
|
|
|
|
|
|
|
logging.basicConfig(level=logging.INFO) |
|
|
logger = logging.getLogger(__name__) |
|
|
|
|
|
|
|
|
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 |
