Create api.py

api.py

@@ -0,0 +1,207 @@
+from fastapi import FastAPI, HTTPException, Depends, status
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+from llama_cpp import Llama
+import os
+import uvicorn
+from typing import Optional, List, Dict, Union, Literal
+import time
+import json
+import uuid
+from datetime import datetime
+
+# Configuration
+VALID_API_KEYS = {
+    "sk-adminkey02",
+    "sk-testkey123",
+    "sk-userkey456",
+    "sk-demokey789"
+}
+
+# Global model variable
+llm = None
+security = HTTPBearer()
+
+class Message(BaseModel):
+    role: Literal["system", "user", "assistant"]
+    content: str
+
+class ChatCompletionRequest(BaseModel):
+    model: str = "zephyr-quiklang-3b-4k"
+    messages: List[Message]
+    max_tokens: Optional[int] = 512
+    temperature: Optional[float] = 0.7
+    top_p: Optional[float] = 0.9
+    n: Optional[int] = 1
+    stream: Optional[bool] = False
+    stop: Optional[Union[str, List[str]]] = None
+
+class ChatCompletionChoice(BaseModel):
+    index: int
+    message: Message
+    finish_reason: Literal["stop", "length", "content_filter"]
+
+class Usage(BaseModel):
+    prompt_tokens: int
+    completion_tokens: int
+    total_tokens: int
+
+class ChatCompletionResponse(BaseModel):
+    id: str
+    object: str = "chat.completion"
+    created: int
+    model: str
+    choices: List[ChatCompletionChoice]
+    usage: Usage
+
+class Model(BaseModel):
+    id: str
+    object: str = "model"
+    created: int
+    owned_by: str
+
+class ModelsResponse(BaseModel):
+    object: str = "list"
+    data: List[Model]
+
+app = FastAPI(
+    title="Zephyr Quiklang OpenAI API",
+    description="OpenAI-compatible API for Zephyr-Quiklang-3B-4K",
+    version="1.0.0",
+    docs_url="/v1/docs",
+    redoc_url="/v1/redoc"
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+def verify_api_key(credentials: HTTPAuthorizationCredentials = Depends(security)):
+    if credentials.credentials not in VALID_API_KEYS:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API key")
+    return credentials.credentials
+
+def load_model():
+    global llm
+    model_path = "zephyr-quiklang-3b-4k.Q4_K_M.gguf"
+
+    if not os.path.exists(model_path):
+        raise Exception(f"Model file {model_path} not found!")
+
+    llm = Llama(
+        model_path=model_path,
+        n_ctx=4096,
+        n_threads=2,
+        n_batch=512,
+        verbose=False,
+        use_mlock=True,
+        n_gpu_layers=0,
+    )
+
+def format_messages(messages: List[Message]) -> str:
+    formatted = ""
+    for message in messages:
+        formatted += f"<|im_start|>{message.role}\n{message.content}\n<|im_end|>\n"
+    formatted += "<|im_start|>assistant\n"
+    return formatted
+
+def count_tokens_rough(text: str) -> int:
+    return len(text.split())
+
+@app.on_event("startup")
+async def startup_event():
+    print("🚀 Starting Zephyr Quiklang API...")
+    load_model()
+    print("✅ Model loaded.")
+
+@app.get("/v1/models", response_model=ModelsResponse)
+async def list_models(api_key: str = Depends(verify_api_key)):
+    return ModelsResponse(data=[
+        Model(
+            id="zephyr-quiklang-3b-4k",
+            created=int(datetime.now().timestamp()),
+            owned_by="local"
+        )
+    ])
+
+@app.post("/v1/chat/completions", response_model=ChatCompletionResponse)
+async def create_chat_completion(request: ChatCompletionRequest, api_key: str = Depends(verify_api_key)):
+    if llm is None:
+        raise HTTPException(status_code=503, detail="Model not loaded")
+
+    prompt = format_messages(request.messages)
+    prompt_tokens = count_tokens_rough(prompt)
+    start_time = time.time()
+
+    try:
+        response = llm(
+            prompt,
+            max_tokens=request.max_tokens,
+            temperature=request.temperature,
+            top_p=request.top_p,
+            stop=["<|im_end|>", "<|im_start|>"] + (request.stop or []),
+            echo=False
+        )
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error generating response: {str(e)}")
+
+    end_time = time.time()
+    generation_time = end_time - start_time
+    response_text = response['choices'][0]['text'].strip()
+    completion_tokens = count_tokens_rough(response_text)
+
+    return ChatCompletionResponse(
+        id=f"chatcmpl-{uuid.uuid4().hex[:8]}",
+        created=int(time.time()),
+        model=request.model,
+        choices=[
+            ChatCompletionChoice(
+                index=0,
+                message=Message(role="assistant", content=response_text),
+                finish_reason="stop"
+            )
+        ],
+        usage=Usage(
+            prompt_tokens=prompt_tokens,
+            completion_tokens=completion_tokens,
+            total_tokens=prompt_tokens + completion_tokens
+        )
+    )
+
+@app.get("/v1/health")
+async def health_check():
+    if llm is None:
+        raise HTTPException(status_code=503, detail="Model not loaded")
+    return {
+        "status": "healthy",
+        "model_loaded": True,
+        "model": "zephyr-quiklang-3b-4k",
+        "timestamp": datetime.now().isoformat()
+    }
+
+@app.get("/v1")
+async def api_info():
+    return {
+        "message": "Zephyr Quiklang OpenAI-Compatible API",
+        "model": "zephyr-quiklang-3b-4k (Q4_K_M)",
+        "endpoints": {
+            "chat_completions": "/v1/chat/completions",
+            "models": "/v1/models",
+            "health": "/v1/health",
+            "docs": "/v1/docs"
+        },
+        "authentication": {
+            "required": True,
+            "type": "Bearer token",
+            "valid_keys": list(VALID_API_KEYS)
+        },
+        "performance": {
+            "context_length": 4096,
+            "expected_speed": "2–8 tok/s (CPU)"
+        }
+    }