Create app.py

app.py

@@ -0,0 +1,181 @@
+"""
+Minecraft Bot LLM Backend - FastAPI Server
+OpenAI-compatible API endpoint with Bearer Token authentication
+"""
+
+import os
+import time
+import uuid
+import logging
+from contextlib import asynccontextmanager
+from typing import List, Optional, Union
+
+from fastapi import FastAPI, HTTPException, Depends, Request
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+import uvicorn
+
+# ─── Logging ───────────────────────────────────────────────────────────────────
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+logger = logging.getLogger(__name__)
+
+# ─── Config ────────────────────────────────────────────────────────────────────
+BEARER_TOKEN = os.environ.get("BEARER_TOKEN", "")
+MODEL_PATH   = os.environ.get("MODEL_PATH", "/app/models/model.gguf")
+MODEL_NAME   = os.environ.get("MODEL_NAME", "qwen2.5-coder-7b-instruct")
+N_CTX        = int(os.environ.get("N_CTX", "4096"))
+N_THREADS    = int(os.environ.get("N_THREADS", "4"))
+N_GPU_LAYERS = int(os.environ.get("N_GPU_LAYERS", "0"))   # 0 = CPU only (HF free tier)
+MAX_TOKENS   = int(os.environ.get("MAX_TOKENS", "1024"))
+
+if not BEARER_TOKEN:
+    logger.warning("⚠️  BEARER_TOKEN is not set — all requests will be REJECTED.")
+
+# ─── Global model holder ────────────────────────────────────────────────────────
+llm = None
+
+# ─── Lifespan: load model once at startup ──────────────────────────────────────
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    global llm
+    logger.info(f"🔄 Loading model from: {MODEL_PATH}")
+    try:
+        from llama_cpp import Llama
+        llm = Llama(
+            model_path=MODEL_PATH,
+            n_ctx=N_CTX,
+            n_threads=N_THREADS,
+            n_gpu_layers=N_GPU_LAYERS,
+            verbose=False,
+            chat_format="chatml",   # works for Qwen2.5 / most instruct models
+        )
+        logger.info("✅ Model loaded successfully.")
+    except Exception as e:
+        logger.error(f"❌ Failed to load model: {e}")
+        raise RuntimeError(f"Model load failed: {e}")
+    yield
+    logger.info("🛑 Shutting down.")
+
+# ─── App ────────────────────────────────────────────────────────────────────────
+app = FastAPI(
+    title="Minecraft Bot LLM API",
+    description="OpenAI-compatible LLM backend for Minecraft automation bot",
+    version="1.0.0",
+    lifespan=lifespan,
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# ─── Auth ────────────────────────────────────────────────────────────────────────
+security = HTTPBearer()
+
+def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
+    if not BEARER_TOKEN:
+        raise HTTPException(status_code=503, detail="Server misconfigured: BEARER_TOKEN not set.")
+    if credentials.credentials != BEARER_TOKEN:
+        raise HTTPException(status_code=401, detail="Invalid or missing Bearer Token.")
+    return credentials.credentials
+
+# ─── Pydantic Schemas (OpenAI-compatible) ──────────────────────────────────────
+class ChatMessage(BaseModel):
+    role: str               # "system" | "user" | "assistant"
+    content: str
+
+class ChatCompletionRequest(BaseModel):
+    model: Optional[str]    = None
+    messages: List[ChatMessage]
+    max_tokens: Optional[int]   = None
+    temperature: Optional[float] = 0.2
+    top_p: Optional[float]       = 0.95
+    stream: Optional[bool]       = False
+    stop: Optional[Union[str, List[str]]] = None
+
+class ChatCompletionChoice(BaseModel):
+    index: int
+    message: ChatMessage
+    finish_reason: str
+
+class UsageInfo(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: UsageInfo
+
+# ─── Routes ─────────────────────────────────────────────────────���───────────────
+@app.get("/")
+async def root():
+    return {"status": "ok", "service": "Minecraft Bot LLM Backend"}
+
+@app.get("/health")
+async def health():
+    return {"status": "healthy", "model_loaded": llm is not None}
+
+@app.get("/v1/models", dependencies=[Depends(verify_token)])
+async def list_models():
+    return {
+        "object": "list",
+        "data": [{"id": MODEL_NAME, "object": "model", "owned_by": "local"}],
+    }
+
+@app.post("/v1/chat/completions", response_model=ChatCompletionResponse, dependencies=[Depends(verify_token)])
+async def chat_completions(request: ChatCompletionRequest):
+    if llm is None:
+        raise HTTPException(status_code=503, detail="Model not loaded yet.")
+
+    messages = [{"role": m.role, "content": m.content} for m in request.messages]
+    max_tok  = request.max_tokens or MAX_TOKENS
+
+    logger.info(f"📨 Inference request | messages={len(messages)} | max_tokens={max_tok}")
+
+    try:
+        result = llm.create_chat_completion(
+            messages=messages,
+            max_tokens=max_tok,
+            temperature=request.temperature,
+            top_p=request.top_p,
+            stop=request.stop or [],
+        )
+    except Exception as e:
+        logger.error(f"Inference error: {e}")
+        raise HTTPException(status_code=500, detail=f"Inference failed: {e}")
+
+    choice  = result["choices"][0]
+    usage   = result.get("usage", {})
+
+    return ChatCompletionResponse(
+        id=f"chatcmpl-{uuid.uuid4().hex}",
+        created=int(time.time()),
+        model=MODEL_NAME,
+        choices=[
+            ChatCompletionChoice(
+                index=0,
+                message=ChatMessage(
+                    role="assistant",
+                    content=choice["message"]["content"],
+                ),
+                finish_reason=choice.get("finish_reason", "stop"),
+            )
+        ],
+        usage=UsageInfo(
+            prompt_tokens=usage.get("prompt_tokens", 0),
+            completion_tokens=usage.get("completion_tokens", 0),
+            total_tokens=usage.get("total_tokens", 0),
+        ),
+    )
+
+# ─── Entry point ────────────────────────────────────────────────────────────────
+if __name__ == "__main__":
+    uvicorn.run("app:app", host="0.0.0.0", port=7860, workers=1)