ai_agents / src /qa_platform /llm_runtime.py
tomas.helmfridsson
Fix HF output limits and requirements feedback routing
d68e71f
Raw History Blame Contribute Delete
19.4 kB
from __future__ import annotations
import json
import os
import socket
import time
from dataclasses import dataclass
from importlib import import_module
from pathlib import Path
from typing import Any
from urllib import error, request
from .models import AgentRuntimeConfig
class LLMRuntimeError(RuntimeError):
pass
HTTP_TIMEOUT_SECONDS = 55
OLLAMA_DISCOVERY_TIMEOUT_SECONDS = 2
HF_OUTPUT_TOKEN_BUDGET = 8192
HF_MODEL_FAMILY_MAP = {
"Qwen 3 32B": "Qwen/Qwen3-32B",
"Qwen3-30B-A3B": "Qwen/Qwen3-30B-A3B",
"Llama 3.3 70B Instruct": "meta-llama/Llama-3.3-70B-Instruct",
"DeepSeek R1": "deepseek-ai/DeepSeek-R1",
"DeepSeek-V3.1": "deepseek-ai/DeepSeek-V3.1",
"gpt-oss-120b": "openai/gpt-oss-120b",
"gpt-oss-20b": "openai/gpt-oss-20b",
}
OLLAMA_MODEL_FAMILY_MAP = {
"Qwen 3 32B": "qwen3:8b",
"Qwen3-30B-A3B": "qwen3:8b",
"Llama 3.3 70B Instruct": "llama3:latest",
"DeepSeek R1": "deepseek-r1:8b",
"DeepSeek-V3.1": "deepseek-r1:8b",
"gpt-oss-120b": "gpt-oss:120b",
"gpt-oss-20b": "gpt-oss:20b",
}
RECOMMENDED_MODEL_FAMILY_BY_AGENT = {
"orchestrator": "Qwen 3 32B",
"requirements_analyst": "Qwen 3 32B",
"test_design": "Llama 3.3 70B Instruct",
"review": "DeepSeek R1",
}
@dataclass
class LLMCallMetadata:
endpoint: str
model_id: str
provider_strategy: str
response_format_used: bool
usage: dict[str, Any]
def load_local_env() -> None:
env_path = Path(__file__).resolve().parents[2] / ".env"
if not env_path.exists():
return
for raw_line in env_path.read_text(encoding="utf-8").splitlines():
line = raw_line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
os.environ.setdefault(key.strip(), value.strip())
def is_llm_enabled(runtime_config: AgentRuntimeConfig | None) -> bool:
return bool(runtime_config and runtime_config.execution_mode == "LLM-backed")
def resolve_live_model_id(runtime_config: AgentRuntimeConfig) -> str:
recommended_family = RECOMMENDED_MODEL_FAMILY_BY_AGENT.get(runtime_config.agent_key, "Qwen 3 32B")
override = (runtime_config.model_override or "").strip()
if override:
return override
family = runtime_config.model_family or recommended_family
strategy = runtime_config.provider_strategy
if strategy == "HF cheapest/free credits":
return f"{HF_MODEL_FAMILY_MAP.get(family, HF_MODEL_FAMILY_MAP[recommended_family])}:cheapest"
if strategy == "HF fastest":
return f"{HF_MODEL_FAMILY_MAP.get(family, HF_MODEL_FAMILY_MAP[recommended_family])}:fastest"
if strategy == "Ollama local":
return OLLAMA_MODEL_FAMILY_MAP.get(family, family)
if strategy == "Custom OpenAI-compatible":
return HF_MODEL_FAMILY_MAP.get(family, HF_MODEL_FAMILY_MAP[recommended_family])
return runtime_config.model_id or HF_MODEL_FAMILY_MAP[recommended_family]
def call_structured_llm(
runtime_config: AgentRuntimeConfig,
system_prompt: str,
user_prompt: str,
schema_name: str,
schema: dict[str, Any],
temperature: float = 0.2,
) -> tuple[dict[str, Any], LLMCallMetadata]:
load_local_env()
if runtime_config.provider_strategy.startswith("HF "):
return _call_hf_structured_llm(
runtime_config=runtime_config,
system_prompt=system_prompt,
user_prompt=user_prompt,
schema_name=schema_name,
schema=schema,
temperature=temperature,
)
endpoint, api_key = _resolve_endpoint_and_key(runtime_config)
_validate_runtime_config(runtime_config)
messages = [
{"role": "system", "content": system_prompt.strip()},
{"role": "user", "content": user_prompt.strip()},
]
response_format = {
"type": "json_schema",
"json_schema": {
"name": schema_name,
"schema": schema,
"strict": True,
},
}
errors: list[str] = []
for model_id in _candidate_model_ids(runtime_config):
try:
payload = _post_chat_completion(
endpoint=endpoint,
api_key=api_key,
model_id=model_id,
messages=messages,
temperature=temperature,
response_format=response_format,
)
content = _extract_message_content(payload)
return json.loads(content), LLMCallMetadata(
endpoint=endpoint,
model_id=model_id,
provider_strategy=runtime_config.provider_strategy,
response_format_used=True,
usage=payload.get("usage") or {},
)
except (LLMRuntimeError, json.JSONDecodeError) as first_error:
errors.append(f"{model_id} structured-output attempt: {first_error}")
fallback_system_prompt = (
system_prompt.strip()
+ "\nReturn only valid JSON matching the requested schema. Do not add markdown fences or commentary."
)
fallback_user_prompt = (
user_prompt.strip()
+ "\nJSON schema to satisfy exactly:\n"
+ json.dumps(schema, ensure_ascii=False)
)
try:
fallback_payload = _post_chat_completion(
endpoint=endpoint,
api_key=api_key,
model_id=model_id,
messages=[
{"role": "system", "content": fallback_system_prompt},
{"role": "user", "content": fallback_user_prompt},
],
temperature=temperature,
response_format=None,
)
content = _extract_message_content(fallback_payload)
return _extract_json_object(content), LLMCallMetadata(
endpoint=endpoint,
model_id=model_id,
provider_strategy=runtime_config.provider_strategy,
response_format_used=False,
usage=fallback_payload.get("usage") or {},
)
except Exception as second_error: # noqa: BLE001
errors.append(f"{model_id} fallback attempt: {second_error}")
raise LLMRuntimeError(
f"Live LLM call failed for {runtime_config.agent_name}. Attempts: {' | '.join(errors)}"
)
def _call_hf_structured_llm(
runtime_config: AgentRuntimeConfig,
system_prompt: str,
user_prompt: str,
schema_name: str,
schema: dict[str, Any],
temperature: float,
) -> tuple[dict[str, Any], LLMCallMetadata]:
endpoint, api_key = _resolve_endpoint_and_key(runtime_config)
model_id = resolve_live_model_id(runtime_config)
# Keep the selected provider strategy throughout retries, and share one budget.
deadline = time.monotonic() + max(0.1, runtime_config.timeout_seconds - 1)
errors: list[str] = []
def remaining_timeout() -> float:
remaining = deadline - time.monotonic()
if remaining <= 0:
raise LLMRuntimeError(
f"HF call budget exhausted for {runtime_config.agent_name}. "
f"Attempts: {' | '.join(errors)}"
)
return remaining
json_messages = [
{
"role": "system",
"content": (
system_prompt.strip()
+ "\nReturn only valid JSON matching the requested schema. "
+ "Do not add markdown fences or commentary."
),
},
{
"role": "user",
"content": (
user_prompt.strip()
+ "\nJSON schema to satisfy exactly:\n"
+ json.dumps({"name": schema_name, "schema": schema}, ensure_ascii=False)
),
},
]
try:
content, usage = _post_hf_inference_chat_completion(
api_key=api_key,
model_id=model_id,
messages=json_messages,
temperature=temperature,
timeout_seconds=remaining_timeout(),
)
return _extract_json_object(content), LLMCallMetadata(
endpoint="huggingface_hub.InferenceClient.chat_completion",
model_id=model_id,
provider_strategy=runtime_config.provider_strategy,
response_format_used=False,
usage=usage,
)
except Exception as exc: # noqa: BLE001
errors.append(f"{model_id} inference-client attempt: {exc}")
response_format = {
"type": "json_schema",
"json_schema": {"name": schema_name, "schema": schema, "strict": True},
}
for format_value in (response_format, None):
timeout_seconds = remaining_timeout()
try:
payload = _post_chat_completion(
endpoint=endpoint,
api_key=api_key,
model_id=model_id,
messages=(
[{"role": "system", "content": system_prompt.strip()},
{"role": "user", "content": user_prompt.strip()}]
if format_value else json_messages
),
temperature=temperature,
response_format=format_value,
timeout_seconds=timeout_seconds,
max_tokens=HF_OUTPUT_TOKEN_BUDGET,
)
if (payload.get("choices") or [{}])[0].get("finish_reason") == "length":
raise LLMRuntimeError("HF response exhausted its 8192-token output budget.")
return _extract_json_object(_extract_message_content(payload)), LLMCallMetadata(
endpoint=endpoint,
model_id=model_id,
provider_strategy=runtime_config.provider_strategy,
response_format_used=bool(format_value),
usage=payload.get("usage") or {},
)
except Exception as exc: # noqa: BLE001
errors.append(f"{model_id} router {'structured' if format_value else 'JSON'} attempt: {exc}")
raise LLMRuntimeError(
f"Live LLM call failed for {runtime_config.agent_name}. Attempts: {' | '.join(errors)}"
)
def _resolve_endpoint_and_key(runtime_config: AgentRuntimeConfig) -> tuple[str, str]:
strategy = runtime_config.provider_strategy
if strategy in {"HF cheapest/free credits", "HF fastest"}:
token = os.environ.get("HF_TOKEN", "").strip()
if not token:
raise LLMRuntimeError("HF_TOKEN is required for Hugging Face live LLM execution.")
return "https://router.huggingface.co/v1/chat/completions", token
if strategy == "Ollama local":
base_url = os.environ.get("OLLAMA_BASE_URL", "http://127.0.0.1:11434").rstrip("/")
return _normalize_chat_endpoint(base_url), os.environ.get("OLLAMA_API_KEY", "ollama")
if strategy == "Custom OpenAI-compatible":
base_url = os.environ.get("OPENAI_BASE_URL", "").strip()
if not base_url:
raise LLMRuntimeError("OPENAI_BASE_URL is required for custom OpenAI-compatible execution.")
return _normalize_chat_endpoint(base_url.rstrip("/")), os.environ.get("OPENAI_API_KEY", "")
raise LLMRuntimeError(f"Unsupported provider strategy: {strategy}")
def _candidate_model_ids(runtime_config: AgentRuntimeConfig) -> list[str]:
resolved = resolve_live_model_id(runtime_config)
candidates = [resolved]
if runtime_config.provider_strategy.startswith("HF ") and ":" in resolved:
base_model = resolved.split(":", 1)[0]
if base_model not in candidates:
candidates.append(base_model)
return candidates
def _validate_runtime_config(runtime_config: AgentRuntimeConfig) -> None:
if runtime_config.provider_strategy != "Ollama local":
return
_assert_ollama_model_available(
base_url=os.environ.get("OLLAMA_BASE_URL", "http://127.0.0.1:11434").rstrip("/"),
model_id=resolve_live_model_id(runtime_config),
)
def _normalize_chat_endpoint(base_url: str) -> str:
if base_url.endswith("/chat/completions"):
return base_url
if base_url.endswith("/v1"):
return f"{base_url}/chat/completions"
return f"{base_url}/v1/chat/completions"
def _ollama_tags_endpoint(base_url: str) -> str:
if base_url.endswith("/v1"):
base_url = base_url[:-3]
if base_url.endswith("/v1/chat/completions"):
base_url = base_url[: -len("/v1/chat/completions")]
return f"{base_url.rstrip('/')}/api/tags"
def _assert_ollama_model_available(base_url: str, model_id: str) -> None:
req = request.Request(_ollama_tags_endpoint(base_url), method="GET")
try:
with request.urlopen(req, timeout=OLLAMA_DISCOVERY_TIMEOUT_SECONDS) as response:
payload = json.loads(response.read().decode("utf-8"))
except error.HTTPError as exc:
detail = exc.read().decode("utf-8", errors="replace")
raise LLMRuntimeError(
"Could not verify local Ollama models before execution. "
f"HTTP {exc.code} from {_ollama_tags_endpoint(base_url)}: {_summarize_error_detail(detail)}"
) from exc
except error.URLError as exc:
raise LLMRuntimeError(
f"Could not reach local Ollama at {base_url} while checking available models: {exc.reason}"
) from exc
except (TimeoutError, socket.timeout) as exc:
raise LLMRuntimeError(
f"Local Ollama at {base_url} did not respond to model discovery within {OLLAMA_DISCOVERY_TIMEOUT_SECONDS} seconds."
) from exc
available_models = {
str(model.get("model") or "").strip()
for model in (payload.get("models") or [])
if isinstance(model, dict)
}
if model_id not in available_models:
available_list = ", ".join(sorted(name for name in available_models if name)) or "inga modeller hittades"
raise LLMRuntimeError(
f"Requested local Ollama model '{model_id}' is not installed at {base_url}. "
f"Available models: {available_list}. Pull the model first with `ollama pull {model_id}`."
)
def _post_hf_inference_chat_completion(
api_key: str,
model_id: str,
messages: list[dict[str, str]],
temperature: float,
timeout_seconds: float = HTTP_TIMEOUT_SECONDS,
) -> tuple[str, dict[str, Any]]:
try:
huggingface_hub = import_module("huggingface_hub")
except ModuleNotFoundError as exc:
raise LLMRuntimeError(
"huggingface_hub is not installed. Install it to use Hugging Face live execution."
) from exc
client = huggingface_hub.InferenceClient(model=model_id, token=api_key, timeout=timeout_seconds)
try:
response = client.chat_completion(
messages=messages,
max_tokens=HF_OUTPUT_TOKEN_BUDGET,
temperature=temperature,
)
except (TimeoutError, socket.timeout) as exc:
raise LLMRuntimeError(
f"Hugging Face InferenceClient timed out after {timeout_seconds:g} seconds."
) from exc
except Exception as exc: # noqa: BLE001
raise LLMRuntimeError(str(exc)) from exc
choice = response.choices[0]
if getattr(choice, "finish_reason", None) == "length":
raise LLMRuntimeError("HF response exhausted its 8192-token output budget.")
message = getattr(choice, "message", None)
content = None
if message is not None:
content = getattr(message, "content", None)
if content is None and isinstance(message, dict):
content = message.get("content")
if not isinstance(content, str) or not content.strip():
raise LLMRuntimeError("Model response did not contain text content.")
usage_obj = getattr(response, "usage", None)
usage = {
"prompt_tokens": getattr(usage_obj, "prompt_tokens", None),
"completion_tokens": getattr(usage_obj, "completion_tokens", None),
"total_tokens": getattr(usage_obj, "total_tokens", None),
}
return content, {key: value for key, value in usage.items() if value is not None}
def _post_chat_completion(
endpoint: str,
api_key: str,
model_id: str,
messages: list[dict[str, str]],
temperature: float,
response_format: dict[str, Any] | None,
timeout_seconds: float = HTTP_TIMEOUT_SECONDS,
max_tokens: int | None = None,
) -> dict[str, Any]:
body: dict[str, Any] = {
"model": model_id,
"messages": messages,
"stream": False,
"temperature": temperature,
}
if max_tokens is not None:
body["max_tokens"] = max_tokens
if response_format:
body["response_format"] = response_format
headers = {
"Content-Type": "application/json",
}
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
encoded = json.dumps(body).encode("utf-8")
req = request.Request(endpoint, data=encoded, headers=headers, method="POST")
try:
with request.urlopen(req, timeout=timeout_seconds) as response:
return json.loads(response.read().decode("utf-8"))
except error.HTTPError as exc:
detail = exc.read().decode("utf-8", errors="replace")
raise LLMRuntimeError(
f"HTTP {exc.code} from model endpoint {endpoint}: {_summarize_error_detail(detail)}"
) from exc
except error.URLError as exc:
raise LLMRuntimeError(f"Could not reach model endpoint {endpoint}: {exc.reason}") from exc
except (TimeoutError, socket.timeout) as exc:
raise LLMRuntimeError(
f"Model endpoint {endpoint} timed out after {timeout_seconds:g} seconds."
) from exc
def _extract_message_content(payload: dict[str, Any]) -> str:
choices = payload.get("choices") or []
if not choices:
raise LLMRuntimeError("Model response did not contain any choices.")
message = choices[0].get("message") or {}
content = message.get("content")
if isinstance(content, str):
return content
if isinstance(content, list):
text_parts = [
part.get("text", "")
for part in content
if isinstance(part, dict) and part.get("type") in {"text", "output_text"}
]
combined = "".join(text_parts).strip()
if combined:
return combined
raise LLMRuntimeError("Model response did not contain text content.")
def _extract_json_object(content: str) -> dict[str, Any]:
stripped = content.strip()
if stripped.startswith("```"):
stripped = stripped.strip("`")
if stripped.startswith("json"):
stripped = stripped[4:].strip()
start = stripped.find("{")
end = stripped.rfind("}")
if start == -1 or end == -1 or end <= start:
raise LLMRuntimeError("Could not locate a JSON object in the model response.")
return json.loads(stripped[start : end + 1])
def _summarize_error_detail(detail: str) -> str:
compact = " ".join(detail.split())
lowered = compact.lower()
if "error 1010" in lowered or "access denied" in lowered:
return "Access denied by the routed provider (Cloudflare 1010). Try another provider strategy or a local/custom backend."
if len(compact) > 320:
return compact[:317] + "..."
return compact