File size: 6,417 Bytes
40e8702 | 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 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 | """
llama.cpp HTTP client wrapper for FormScout.
Wraps the llama.cpp server's /completion and /embedding endpoints.
Falls back gracefully when the server is unavailable.
Model: Qwen3-VL-8B-Instruct (Q4_K_M GGUF) for VLM inference.
Model: Qwen3-VL-Embedding-8B (Q4_K_M GGUF) for embeddings.
Params: 8B each (shared backbone).
License: Apache-2.0.
"""
from __future__ import annotations
import base64
import json
import logging
from pathlib import Path
from typing import Any
import requests
from formscout import config
logger = logging.getLogger(__name__)
_TIMEOUT = 120 # seconds — VLM can be slow
class LlamaCppClient:
"""HTTP client for a llama.cpp server instance."""
def __init__(self, host: str | None = None, port: int | None = None):
self.host = host or config.LLAMA_CPP_HOST
self.port = port or config.LLAMA_CPP_PORT_VLM
self.base_url = f"http://{self.host}:{self.port}"
@property
def available(self) -> bool:
"""Check if the server is reachable."""
try:
r = requests.get(f"{self.base_url}/health", timeout=5)
return r.status_code == 200
except (requests.ConnectionError, requests.Timeout):
return False
def complete(
self,
prompt: str,
images: list[str] | None = None,
max_tokens: int = 512,
temperature: float = 0.1,
stop: list[str] | None = None,
) -> dict[str, Any]:
"""
Send a chat-completion request (OpenAI-compatible /v1/chat/completions —
required for multimodal: llama-server routes images through the mmproj
only on this endpoint). Returns parsed JSON if the response is JSON,
otherwise returns {"text": raw_text}.
Args:
prompt: The text prompt (system + user combined).
images: Optional list of base64-encoded JPEGs or file paths.
max_tokens: Max generation tokens.
temperature: Sampling temperature.
stop: Stop sequences (default: none — JSON output must not be truncated).
"""
content: list[dict[str, Any]] = [{"type": "text", "text": prompt}]
for img in images or []:
if len(img) < 4096 and Path(img).exists():
with open(img, "rb") as f:
b64 = base64.b64encode(f.read()).decode()
else:
b64 = img # already base64
content.append({
"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{b64}"},
})
payload: dict[str, Any] = {
"model": config.LLAMA_CPP_MODEL,
"messages": [{"role": "user", "content": content}],
"max_tokens": max_tokens,
"temperature": temperature,
}
if stop:
payload["stop"] = stop
result = self._post(payload)
if "error" in result and images:
# Multimodal failed — retry text-only so scoring still proceeds.
logger.warning("Multimodal request failed (%s), retrying text-only", result["error"])
text_payload = {
"model": config.LLAMA_CPP_MODEL,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": max_tokens,
"temperature": temperature,
}
if stop:
text_payload["stop"] = stop
result = self._post(text_payload)
return result
def _post(self, payload: dict[str, Any]) -> dict[str, Any]:
"""POST a chat-completion payload, surfacing the response body on errors."""
try:
r = requests.post(
f"{self.base_url}/v1/chat/completions",
json=payload,
timeout=_TIMEOUT,
)
if not r.ok:
# Capture the server's explanation (e.g. "Invalid image ...")
body = ""
try:
body = r.text[:500]
except Exception:
pass
logger.warning("llama-server %s: %s", r.status_code, body)
return {"error": f"HTTP {r.status_code}: {body}", "text": ""}
result = r.json()
text = result["choices"][0]["message"]["content"] or ""
return self._parse_json_reply(text)
except requests.ConnectionError:
return {"error": "llama.cpp server not available", "text": ""}
except requests.Timeout:
return {"error": "llama.cpp server timeout", "text": ""}
except Exception as e:
return {"error": str(e), "text": ""}
@staticmethod
def _parse_json_reply(text: str) -> dict[str, Any]:
"""Parse model output as JSON, tolerating markdown fences."""
stripped = text.strip()
if stripped.startswith("```"):
stripped = stripped.split("\n", 1)[-1]
stripped = stripped.rsplit("```", 1)[0].strip()
try:
parsed = json.loads(stripped)
if isinstance(parsed, dict):
return parsed
except (json.JSONDecodeError, TypeError):
pass
return {"text": text}
class EmbeddingClient:
"""HTTP client for the llama.cpp embedding server."""
def __init__(self, host: str | None = None, port: int | None = None):
self.host = host or config.LLAMA_CPP_HOST
self.port = port or config.LLAMA_CPP_PORT_EMBED
self.base_url = f"http://{self.host}:{self.port}"
@property
def available(self) -> bool:
try:
r = requests.get(f"{self.base_url}/health", timeout=5)
return r.status_code == 200
except (requests.ConnectionError, requests.Timeout):
return False
def embed(self, text: str) -> list[float] | None:
"""Get embedding vector for text. Returns None on failure."""
try:
r = requests.post(
f"{self.base_url}/embedding",
json={"content": text},
timeout=30,
)
r.raise_for_status()
data = r.json()
return data.get("embedding")
except Exception:
return None
|