""" llama_cpp_server_proxy.py — Proxy llama.cpp API server on port 8080. Proxies to Ollama running on port 11434. Provides OpenAI-compatible API endpoints expected by LlamaCppClient in CyberBrain26. Install dependencies: pip install fastapi uvicorn httpx Run: python3 ./llama_cpp_server_proxy.py """ import json import logging import os import re import sys import time import httpx from fastapi import FastAPI, Request from fastapi.responses import JSONResponse, StreamingResponse logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s [%(name)s] %(message)s") log = logging.getLogger("llama_cpp_server_proxy") OLLAMA_BASE_URL = os.environ.get("OLLAMA_BASE_URL", "http://127.0.0.1:11434") PRIMARY_MODEL = os.environ.get("PRIMARY_MODEL", "phi3.5:latest") FALLBACK_MODEL = os.environ.get("FALLBACK_MODEL", "dolphin-cb26:latest") PORT = int(os.environ.get("LLM_PROXY_PORT", "8080")) HOST = os.environ.get("LLM_PROXY_HOST", "127.0.0.1") app = FastAPI() @app.get("/") async def root(): return { "os": sys.platform, "arch": "llama.cpp proxy", "model": PRIMARY_MODEL, "port": PORT, } @app.get("/api/tokenize") async def list_models(): try: async with httpx.AsyncClient(timeout=30) as client: resp = await client.get(f"{OLLAMA_BASE_URL}/api/tags") if resp.status_code == 200: data = resp.json() model_info = [] for m in data.get("models", []): model_info.append({ "model": m["name"], "context_length": m.get("details", {}).get("context_length", 0), }) return {"models": model_info} return {"models": [], "message": "error"} except Exception as e: # noqa: BLE001 log.error(f"list_models error: {e}") return {"models": [], "message": str(e)} @app.post("/api/chat") async def chat_completions(request: Request): body = await request.json() if request.headers.get("content-type", "").startswith("application/json") else {} model = body.get("model") or PRIMARY_MODEL messages = body.get("messages", []) stream = body.get("stream", False) options = body.get("options", {}) # Determine primary or fallback model primary_ok = False try: async with httpx.AsyncClient(timeout=30) as client: resp = await client.post(f"{OLLAMA_BASE_URL}/api/show", json={"name": model}) primary_ok = resp.status_code in (200,) except httpx.HTTPError as e: log.warning(f"Primary model check HTTP error: {e}") if not primary_ok: fallback = FALLBACK_MODEL if ":" not in FALLBACK_MODEL else model log.info(f"Primary model '{model}' unavailable, falling back to '{fallback}'") model = fallback body = { "model": model, "messages": messages, "stream": stream, "options": options, } try: async with httpx.AsyncClient(timeout=600) as client: if stream: return StreamingResponse(StreamChatIterator(client, body)) else: resp = await client.post(f"{OLLAMA_BASE_URL}/api/chat", json=body) if resp.status_code == 200: data = resp.json() choices = data.get("message", {}).get("content", "") if isinstance(data.get("message"), dict) else str(data.get("message", "")) return { "id": f"chatcmpl-{int(time.time())}", "object": "chat.completion", "created": int(time.time()), "model": model, "choices": [{"index": 0, "message": {"role": "assistant", "content": choices}, "finish_reason": "stop"}], "usage": { "prompt_tokens": sum(len(msg.get("content", "")) for msg in messages), "completion_tokens": len(choices), "total_tokens": sum(len(msg.get("content", "")) for msg in messages) + len(choices), } } else: return JSONResponse({"error": f"Ollama error {resp.status_code}: {resp.text}"}, status_code=500) except httpx.HTTPError as e: log.error(f"chat HTTP error: {e}") return JSONResponse({"error": f"chat http {e.__class__.__name__}"}, status_code=500) except Exception as e: # noqa: BLE001 log.error(f"chat error: {e}") return JSONResponse({"error": str(e)}, status_code=500) class StreamChatIterator: def __init__(self, client, body): self.client = client self.body = body async def __iter__(self): request_url = "/api/chat" response = await self.client.post(f"{OLLAMA_BASE_URL}{request_url}", json=self.body) for line in response.iter_lines(): line = line.strip() if not line or not line.startswith("{") or not line.endswith("}"): continue try: chunk = json.loads(line) yield {"model": chunk.get("model", ""), "content": chunk.get("message", {}).get("content", ""), "done": chunk.get("done", False)} except json.JSONDecodeError: continue @app.post("/api/completion") async def completion_completions(request: Request): body = await request.json() if request.headers.get("content-type", "").startswith("application/json") else {} model = body.get("model") or PRIMARY_MODEL prompt = body.get("prompt", "") stream = body.get("stream", False) options = body.get("options", {}) try: async with httpx.AsyncClient(timeout=600) as client: if stream: return StreamingResponse(StreamCompletionIterator(client, model, prompt, options)) else: resp = await client.post(f"{OLLAMA_BASE_URL}/api/generate", json={"model": model, "prompt": prompt, "stream": False, **options}) if resp.status_code == 200: data = resp.json() return { "id": f"cmpl-{int(time.time())}", "object": "text_completion", "created": int(time.time()), "model": model, "choices": [{"index": 0, "text": data.get("response", ""), "finish_reason": "stop"}], } return JSONResponse({"error": f"Ollama error {resp.status_code}: {resp.text}"}, status_code=500) except httpx.HTTPError as e: log.error(f"completion HTTP error: {e}") return JSONResponse({"error": f"completion http {e.__class__.__name__}"}, status_code=500) except Exception as e: # noqa: BLE001 log.error(f"completion error: {e}") return JSONResponse({"error": str(e)}, status_code=500) class StreamCompletionIterator: def __init__(self, client, model, prompt, options): self.client = client self.model = model self.prompt = prompt self.options = options async def __iter__(self): response = await self.client.post(f"{OLLAMA_BASE_URL}/api/generate", json={"model": self.model, "prompt": self.prompt, "stream": True, **self.options}) for line in response.iter_lines(): line = line.strip() if not line or not line.startswith("{"): continue try: chunk = json.loads(line) text = chunk.get("response", "") if text: yield {"model": self.model, "text": text, "done": chunk.get("done", False)} except json.JSONDecodeError: continue @app.post("/api/embeddings") async def embeddings(request: Request): body = await request.json() if request.headers.get("content-type", "").startswith("application/json") else {} model = body.get("model") or FALLBACK_MODEL input_text = body.get("input", []) try: async with httpx.AsyncClient(timeout=60) as client: if isinstance(input_text, str): prompt = input_text else: prompt = " ".join(input_text) resp = await client.post(f"{OLLAMA_BASE_URL}/api/embed", json={"model": model, "prompt": prompt}) if resp.status_code == 200: data = resp.json() embeddings = data.get("embeddings", []) return {"embeddings": embeddings} return JSONResponse({"error": f"Ollama error {resp.status_code}: {resp.text}"}, status_code=500) except httpx.HTTPError as e: log.error(f"embeddings HTTP error: {e}") return JSONResponse({"error": f"embeddings http {e.__class__.__name__}"}, status_code=500) except Exception as e: # noqa: BLE001 log.error(f"embeddings error: {e}") return JSONResponse({"error": str(e)}, status_code=500) @app.post("/api/tokenize") async def tokenize(request: Request): # Ollama does not expose a public tokenization API; /api/generate returns # generated text, not token IDs, so regex extraction below is unreliable. # Return 501 rather than silently producing incorrect tokens. return JSONResponse( {"error": "Tokenization not supported via Ollama proxy"}, status_code=501 ) @app.post("/api/detokenize") async def detokenize(request: Request): body = await request.json() if request.headers.get("content-type", "").startswith("application/json") else {} model = body.get("model") or FALLBACK_MODEL tokens = body.get("tokens", []) token_str = " ".join(str(t) for t in tokens) if isinstance(tokens, list) else str(tokens) try: async with httpx.AsyncClient(timeout=30) as client: resp = await client.post(f"{OLLAMA_BASE_URL}/api/generate", json={"model": model, "prompt": token_str, "stream": False, "format": "json"}) if resp.status_code == 200: data = resp.json() text = data.get("response", "") printable = re.sub(r'[^\x20-\x7E\n]', '', text) return {"text": printable} return JSONResponse({"error": f"Ollama error {resp.status_code}: {resp.text}"}, status_code=500) except Exception as e: # noqa: BLE001 log.error(f"detokenize error: {e}") return JSONResponse({"error": str(e)}, status_code=500) if __name__ == "__main__": import uvicorn uvicorn.run(app, host=HOST, port=PORT, log_level="info")