"""Reusable LLM client with prompt templates, response parsing, and retry logic.
Provides a unified interface for LLM calls across the application.
Configured via environment variables:
- BUILDER_LLM_HOST: LLM server URL (default: http://localhost:8086)
- LLM_API_KEY: API key (optional, for OpenAI-compatible servers)
- LLM_MODEL: Model name (default: qwen3.5-4b)
Falls back gracefully to template-based suggestions when LLM is unavailable.
"""
import json
import logging
import os
from typing import Any, Optional
logger = logging.getLogger("agentforms.llm")
# ─── Configuration ──────────────────────────────────────────────────────────────
def _get_config() -> dict:
"""Get LLM configuration from environment variables."""
return {
"host": os.environ.get("BUILDER_LLM_HOST", "http://localhost:8086"),
"api_key": os.environ.get("LLM_API_KEY", ""),
"model": os.environ.get("LLM_MODEL", "qwen3.5-4b"),
"temperature": float(os.environ.get("LLM_TEMPERATURE", "0.3")),
"max_tokens": int(os.environ.get("LLM_MAX_TOKENS", "2048")),
"timeout": int(os.environ.get("LLM_TIMEOUT", "30")),
}
# ─── LLM Client ─────────────────────────────────────────────────────────────────
class LLMClient:
"""OpenAI-compatible LLM client with retry and JSON parsing."""
def __init__(self, config: dict | None = None):
self.config = config or _get_config()
def generate(
self,
prompt: str,
system_prompt: str = "",
temperature: float | None = None,
max_tokens: int | None = None,
json_mode: bool = False,
) -> dict:
"""Generate text from the LLM.
Args:
prompt: User message
system_prompt: System message for context
temperature: Sampling temperature (default from config)
max_tokens: Max output tokens (default from config)
json_mode: If True, enforce JSON response and parse it
Returns:
{
"success": bool,
"text": str, # Raw LLM output
"parsed": Any, # Parsed JSON if json_mode=True
"error": str|None, # Error message on failure
"model": str, # Model used
}
"""
import requests
cfg = self.config
temperature = temperature if temperature is not None else cfg["temperature"]
max_tokens = max_tokens if max_tokens is not None else cfg["max_tokens"]
# Build messages
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
# If json_mode, append JSON instruction to system prompt
if json_mode and system_prompt:
system_prompt += (
"\n\nReturn ONLY valid JSON matching the specified schema. "
"Do not include markdown code fences or explanatory text."
)
messages[0]["content"] = system_prompt
elif json_mode:
messages.insert(
0,
{
"role": "system",
"content": "You are a JSON-only API. Return valid JSON with no markdown or explanations.",
},
)
messages.append({"role": "user", "content": prompt})
# Build request
payload = {
"model": cfg["model"],
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
}
# Add response format if json_mode
if json_mode:
payload["response_format"] = {"type": "json_object"}
# Build headers
headers = {"Content-Type": "application/json"}
if cfg["api_key"]:
headers["Authorization"] = f"Bearer {cfg['api_key']}"
try:
url = f"{cfg['host'].rstrip('/')}/v1/chat/completions"
resp = requests.post(
url,
json=payload,
headers=headers,
timeout=cfg["timeout"],
)
resp.raise_for_status()
data = resp.json()
# Extract text from response
text = data["choices"][0]["message"]["content"].strip()
result = {
"success": True,
"text": text,
"parsed": None,
"error": None,
"model": data.get("model", cfg["model"]),
}
# Parse JSON if requested
if json_mode:
result["parsed"] = _parse_json(text)
if result["parsed"] is None:
result["success"] = False
result["error"] = "LLM returned invalid JSON"
return result
except requests.exceptions.Timeout:
logger.error("LLM request timed out after %ds", cfg["timeout"])
return {
"success": False,
"text": "",
"parsed": None,
"error": f"LLM timeout after {cfg['timeout']}s",
"model": cfg["model"],
}
except requests.exceptions.ConnectionError:
logger.error("LLM connection failed: %s", cfg["host"])
return {
"success": False,
"text": "",
"parsed": None,
"error": f"LLM connection failed: {cfg['host']}",
"model": cfg["model"],
}
except Exception as e:
logger.error("LLM request failed: %s", e)
return {
"success": False,
"text": "",
"parsed": None,
"error": str(e),
"model": cfg["model"],
}
def generate_with_retry(
self,
prompt: str,
system_prompt: str = "",
max_retries: int = 2,
json_mode: bool = False,
) -> dict:
"""Generate with automatic retry on parse failure.
Useful for JSON responses that might need a second attempt.
"""
result = self.generate(
prompt=prompt,
system_prompt=system_prompt,
json_mode=json_mode,
)
if not json_mode:
return result
# Retry on JSON parse failure
attempts = 1
while not result["success"] and attempts < max_retries:
logger.info("LLM retry %d/%d for JSON response", attempts + 1, max_retries)
# Lower temperature for more deterministic retry
result = self.generate(
prompt=prompt,
system_prompt=system_prompt,
temperature=0.1,
json_mode=json_mode,
)
attempts += 1
return result
# ─── JSON Parsing ───────────────────────────────────────────────────────────────
def _parse_json(text: str) -> Any:
"""Parse JSON from LLM output, handling common formatting issues.
Handles:
- Raw JSON
- Markdown code fences (```json ... ```)
- Extra whitespace
- Trailing commas (lenient parsing)
"""
if not text:
return None
# Strip markdown code fences
text = text.strip()
if text.startswith("```"):
# Remove opening fence
text = text.split("\n", 1)[-1] if "\n" in text else text[3:]
# Remove closing fence
if text.endswith("```"):
text = text[:-3]
text = text.strip()
# Try strict JSON first
try:
return json.loads(text)
except json.JSONDecodeError:
pass
# Try lenient parsing (remove trailing commas)
try:
import re
# Remove trailing commas before closing brackets/braces
lenient = re.sub(r",\s*([}\]])", r"\1", text)
return json.loads(lenient)
except Exception:
pass
return None
# ─── Global Client ──────────────────────────────────────────────────────────────
# Module-level client for convenience
_client = None
def get_client() -> LLMClient:
"""Get the global LLM client instance."""
global _client
if _client is None:
_client = LLMClient()
return _client
def suggest_chains_llm(
prompt: str,
fields: list,
form_type: str = "form",
) -> dict:
"""LLM-powered chain suggestion for a form.
This is the LLM path for chain_suggester.py. Falls back to
template-based suggestions when the LLM is unavailable.
Args:
prompt: Natural language description of the form purpose
fields: List of field definitions
form_type: Type of form
Returns:
{
"success": bool,
"suggestions": list, # Chain suggestions
"source": str, # "llm" or "template_fallback"
"error": str|None,
}
"""
# Build system prompt
from app.services.chain_suggester import CHAIN_STEP_TYPES
from app.services.chain_templates import CHAIN_TEMPLATES
system_prompt = (
"You are an automated workflow suggestion engine for AgentForms, "
"a form automation platform.\n\n"
"Given a user's form description and form fields, suggest optimal "
"post-submission action chains.\n\n"
"Available action step types:\n" + "\n".join(f"- {t}" for t in sorted(CHAIN_STEP_TYPES)) + "\n\n"
"Available pre-built chain templates by category:\n"
+ "\n".join(f"- {v.get('category', 'general')}: {v.get('name', 'unnamed')}" for v in CHAIN_TEMPLATES.values())
+ "\n\n"
"Rules:\n"
"1. Always suggest the MOST relevant chain based on the form purpose\n"
"2. Fill in realistic defaults for config fields when possible\n"
"3. Mark fields that require user configuration with 'requires_config'\n"
"4. Use {{field_key}} template syntax for referencing form fields\n"
"5. Keep chains practical (2-6 steps, not excessive)\n"
"6. Confidence: high (0.8+) when form purpose clearly matches a template\n\n"
"Response format — return ONLY valid JSON:\n"
"{\n"
' "suggested_chains": [\n'
" {\n"
' "name": "string",\n'
' "description": "string",\n'
' "confidence": 0.0-1.0,\n'
' "category": "string",\n'
' "steps": [\n'
" {\n"
' "type": "string",\n'
' "name": "string",\n'
' "config": {},\n'
' "requires_config": ["field_names"]\n'
" }\n"
" ]\n"
" }\n"
" ]\n"
"}"
)
# Build field list
field_list = json.dumps(
[
{
"key": f.get("key", f.get("name", "")),
"type": f.get("type", "text"),
"label": f.get("label", ""),
}
for f in fields
]
)
# Build user prompt
user_prompt = (
f"Form purpose: {prompt}\n"
f"Form type: {form_type}\n\n"
f"Form fields:\n{field_list}\n\n"
"Suggest the best post-submission action chain(s) for this form.\n"
"Consider: What typically happens after someone submits this kind of form?\n"
"Who needs to be notified? Where does the data need to go?"
)
# Call LLM
client = get_client()
result = client.generate_with_retry(
prompt=user_prompt,
system_prompt=system_prompt,
max_retries=2,
json_mode=True,
)
if result["success"] and result["parsed"]:
suggestions = result["parsed"].get("suggested_chains", [])
return {
"success": True,
"suggestions": suggestions,
"source": "llm",
"error": None,
}
# LLM failed — fall back to template-based suggestions
logger.warning("LLM suggestion failed (%s), using template fallback", result.get("error"))
from app.services.chain_suggester import suggest_chains as template_suggest
suggestions = template_suggest(prompt, fields, form_type)
return {
"success": True,
"suggestions": suggestions,
"source": "template_fallback",
"error": None,
}
def is_llm_available() -> bool:
"""Check if the LLM server is reachable."""
client = get_client()
result = client.generate(
prompt="Respond with only: OK",
temperature=0,
max_tokens=5,
)
return result["success"]