File size: 2,615 Bytes
a2d424a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 |
"""
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 |