import os
from huggingface_hub import hf_hub_download
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from typing import List, Optional
import uvicorn
# -------------------- Configuration --------------------
MODEL_REPO = "Qwen/Qwen2.5-1.5B-Instruct-GGUF"
MODEL_FILE = "qwen2.5-1.5b-instruct-q4_k_m.gguf"
MODEL_PATH = "/data/model.gguf"
PORT = int(os.environ.get("PORT", 7860))
# -------------------- Download model if not present --------------------
def download_model():
if os.path.exists(MODEL_PATH) and os.path.getsize(MODEL_PATH) > 1_000_000:
print("Model already downloaded.")
return
print(f"Downloading {MODEL_FILE} from {MODEL_REPO} ...")
downloaded_path = hf_hub_download(
repo_id=MODEL_REPO,
filename=MODEL_FILE,
local_dir="/data",
local_dir_use_symlinks=False,
)
# Rename to standard path if needed
if downloaded_path != MODEL_PATH:
os.rename(downloaded_path, MODEL_PATH)
print("Download complete.")
download_model()
from llama_cpp import Llama
print("Loading model...")
llm = Llama(
model_path=MODEL_PATH,
n_ctx=2048,
n_threads=2, # HF free tier = 2 vCPUs
verbose=False,
)
print("Model ready.")
# -------------------- FastAPI setup --------------------
app = FastAPI(title="Eudora Orchestrator")
# Embedded HTML UI
html_content = '''
Eudora Orchestrator Tester
🧠Eudora Orchestrator – Chat UI
Response:
Waiting...
'''
class Message(BaseModel):
role: str
content: str
class ChatCompletionRequest(BaseModel):
messages: List[Message]
temperature: Optional[float] = 0.7
max_tokens: Optional[int] = 256
class ChatCompletionResponse(BaseModel):
choices: List[dict]
@app.get("/")
async def root():
return HTMLResponse(content=html_content)
@app.post("/v1/chat/completions", response_model=ChatCompletionResponse)
async def chat_completions(request: ChatCompletionRequest):
msgs = [{"role": m.role, "content": m.content} for m in request.messages]
try:
output = llm.create_chat_completion(
messages=msgs,
temperature=request.temperature,
max_tokens=request.max_tokens,
)
assistant_msg = output["choices"][0]["message"]["content"]
return ChatCompletionResponse(
choices=[{"message": {"role": "assistant", "content": assistant_msg}}]
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/health")
async def health():
return {"status": "ok", "model_loaded": True}
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=PORT)