201 lines
7.3 KiB
Python
201 lines
7.3 KiB
Python
# [DEF:backend.src.services.llm_prompt_templates:Module]
|
|
# @TIER: STANDARD
|
|
# @SEMANTICS: llm, prompts, templates, settings
|
|
# @PURPOSE: Provide default LLM prompt templates and normalization helpers for runtime usage.
|
|
# @LAYER: Domain
|
|
# @RELATION: DEPENDS_ON -> backend.src.core.config_manager
|
|
# @INVARIANT: All required prompt template keys are always present after normalization.
|
|
|
|
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
from typing import Dict, Any, Optional
|
|
|
|
|
|
# [DEF:DEFAULT_LLM_PROMPTS:Constant]
|
|
# @TIER: STANDARD
|
|
# @PURPOSE: Default prompt templates used by documentation, dashboard validation, and git commit generation.
|
|
DEFAULT_LLM_PROMPTS: Dict[str, str] = {
|
|
"dashboard_validation_prompt": (
|
|
"Analyze the attached dashboard screenshot and the following execution logs for health and visual issues.\n\n"
|
|
"Logs:\n"
|
|
"{logs}\n\n"
|
|
"Provide the analysis in JSON format with the following structure:\n"
|
|
"{\n"
|
|
' "status": "PASS" | "WARN" | "FAIL",\n'
|
|
' "summary": "Short summary of findings",\n'
|
|
' "issues": [\n'
|
|
" {\n"
|
|
' "severity": "WARN" | "FAIL",\n'
|
|
' "message": "Description of the issue",\n'
|
|
' "location": "Optional location info (e.g. chart name)"\n'
|
|
" }\n"
|
|
" ]\n"
|
|
"}"
|
|
),
|
|
"documentation_prompt": (
|
|
"Generate professional documentation for the following dataset and its columns.\n"
|
|
"Dataset: {dataset_name}\n"
|
|
"Columns: {columns_json}\n\n"
|
|
"Provide the documentation in JSON format:\n"
|
|
"{\n"
|
|
' "dataset_description": "General description of the dataset",\n'
|
|
' "column_descriptions": [\n'
|
|
" {\n"
|
|
' "name": "column_name",\n'
|
|
' "description": "Generated description"\n'
|
|
" }\n"
|
|
" ]\n"
|
|
"}"
|
|
),
|
|
"git_commit_prompt": (
|
|
"Generate a concise and professional git commit message based on the following diff and recent history.\n"
|
|
"Use Conventional Commits format (e.g., feat: ..., fix: ..., docs: ...).\n\n"
|
|
"Recent History:\n"
|
|
"{history}\n\n"
|
|
"Diff:\n"
|
|
"{diff}\n\n"
|
|
"Commit Message:"
|
|
),
|
|
}
|
|
# [/DEF:DEFAULT_LLM_PROMPTS:Constant]
|
|
|
|
|
|
# [DEF:DEFAULT_LLM_PROVIDER_BINDINGS:Constant]
|
|
# @TIER: STANDARD
|
|
# @PURPOSE: Default provider binding per task domain.
|
|
DEFAULT_LLM_PROVIDER_BINDINGS: Dict[str, str] = {
|
|
"dashboard_validation": "",
|
|
"documentation": "",
|
|
"git_commit": "",
|
|
}
|
|
# [/DEF:DEFAULT_LLM_PROVIDER_BINDINGS:Constant]
|
|
|
|
|
|
# [DEF:DEFAULT_LLM_ASSISTANT_SETTINGS:Constant]
|
|
# @TIER: STANDARD
|
|
# @PURPOSE: Default planner settings for assistant chat intent model/provider resolution.
|
|
DEFAULT_LLM_ASSISTANT_SETTINGS: Dict[str, str] = {
|
|
"assistant_planner_provider": "",
|
|
"assistant_planner_model": "",
|
|
}
|
|
# [/DEF:DEFAULT_LLM_ASSISTANT_SETTINGS:Constant]
|
|
|
|
|
|
# [DEF:normalize_llm_settings:Function]
|
|
# @TIER: STANDARD
|
|
# @PURPOSE: Ensure llm settings contain stable schema with prompts section and default templates.
|
|
# @PRE: llm_settings is dictionary-like value or None.
|
|
# @POST: Returned dict contains prompts with all required template keys.
|
|
def normalize_llm_settings(llm_settings: Any) -> Dict[str, Any]:
|
|
normalized: Dict[str, Any] = {
|
|
"providers": [],
|
|
"default_provider": "",
|
|
"prompts": {},
|
|
"provider_bindings": {},
|
|
**DEFAULT_LLM_ASSISTANT_SETTINGS,
|
|
}
|
|
if isinstance(llm_settings, dict):
|
|
normalized.update(
|
|
{
|
|
k: v
|
|
for k, v in llm_settings.items()
|
|
if k
|
|
in (
|
|
"providers",
|
|
"default_provider",
|
|
"prompts",
|
|
"provider_bindings",
|
|
"assistant_planner_provider",
|
|
"assistant_planner_model",
|
|
)
|
|
}
|
|
)
|
|
prompts = normalized.get("prompts") if isinstance(normalized.get("prompts"), dict) else {}
|
|
merged_prompts = deepcopy(DEFAULT_LLM_PROMPTS)
|
|
merged_prompts.update({k: v for k, v in prompts.items() if isinstance(v, str) and v.strip()})
|
|
normalized["prompts"] = merged_prompts
|
|
bindings = normalized.get("provider_bindings") if isinstance(normalized.get("provider_bindings"), dict) else {}
|
|
merged_bindings = deepcopy(DEFAULT_LLM_PROVIDER_BINDINGS)
|
|
merged_bindings.update({k: v for k, v in bindings.items() if isinstance(v, str)})
|
|
normalized["provider_bindings"] = merged_bindings
|
|
for key, default_value in DEFAULT_LLM_ASSISTANT_SETTINGS.items():
|
|
value = normalized.get(key, default_value)
|
|
normalized[key] = value.strip() if isinstance(value, str) else default_value
|
|
return normalized
|
|
# [/DEF:normalize_llm_settings:Function]
|
|
|
|
|
|
# [DEF:is_multimodal_model:Function]
|
|
# @TIER: STANDARD
|
|
# @PURPOSE: Heuristically determine whether model supports image input required for dashboard validation.
|
|
# @PRE: model_name may be empty or mixed-case.
|
|
# @POST: Returns True when model likely supports multimodal input.
|
|
def is_multimodal_model(model_name: str, provider_type: Optional[str] = None) -> bool:
|
|
token = (model_name or "").strip().lower()
|
|
if not token:
|
|
return False
|
|
provider = (provider_type or "").strip().lower()
|
|
text_only_markers = (
|
|
"text-only",
|
|
"embedding",
|
|
"rerank",
|
|
"whisper",
|
|
"tts",
|
|
"transcribe",
|
|
)
|
|
if any(marker in token for marker in text_only_markers):
|
|
return False
|
|
multimodal_markers = (
|
|
"gpt-4o",
|
|
"gpt-4.1",
|
|
"vision",
|
|
"vl",
|
|
"gemini",
|
|
"claude-3",
|
|
"claude-sonnet-4",
|
|
"omni",
|
|
"multimodal",
|
|
"pixtral",
|
|
"llava",
|
|
"internvl",
|
|
"qwen-vl",
|
|
"qwen2-vl",
|
|
)
|
|
if any(marker in token for marker in multimodal_markers):
|
|
return True
|
|
return False
|
|
# [/DEF:is_multimodal_model:Function]
|
|
|
|
|
|
# [DEF:resolve_bound_provider_id:Function]
|
|
# @TIER: STANDARD
|
|
# @PURPOSE: Resolve provider id configured for a task binding with fallback to default provider.
|
|
# @PRE: llm_settings is normalized or raw dict from config.
|
|
# @POST: Returns configured provider id or fallback id/empty string when not defined.
|
|
def resolve_bound_provider_id(llm_settings: Any, task_key: str) -> str:
|
|
normalized = normalize_llm_settings(llm_settings)
|
|
bindings = normalized.get("provider_bindings", {})
|
|
bound = bindings.get(task_key)
|
|
if isinstance(bound, str) and bound.strip():
|
|
return bound.strip()
|
|
default_provider = normalized.get("default_provider", "")
|
|
return default_provider.strip() if isinstance(default_provider, str) else ""
|
|
# [/DEF:resolve_bound_provider_id:Function]
|
|
|
|
|
|
# [DEF:render_prompt:Function]
|
|
# @TIER: STANDARD
|
|
# @PURPOSE: Render prompt template using deterministic placeholder replacement with graceful fallback.
|
|
# @PRE: template is a string and variables values are already stringifiable.
|
|
# @POST: Returns rendered prompt text with known placeholders substituted.
|
|
def render_prompt(template: str, variables: Dict[str, Any]) -> str:
|
|
rendered = template
|
|
for key, value in variables.items():
|
|
rendered = rendered.replace("{" + key + "}", str(value))
|
|
return rendered
|
|
# [/DEF:render_prompt:Function]
|
|
|
|
|
|
# [/DEF:backend.src.services.llm_prompt_templates:Module]
|