"""Chain Suggester — recommend action chains based on form fields and user intent.
Analyzes a user's natural language prompt + form field definitions to suggest
the best matching pre-built chain templates. In fallback mode (use_fallback=True),
uses keyword matching without requiring an LLM.
Example:
>>> suggest_chains("Send me an email when someone submits", fields)
[
{
"name": "Submission Alert",
"description": "Email the submitter a confirmation, notify admins",
"steps": [...],
"confidence": 0.85,
"requires_config": ["webhook_url"],
"source": "template"
}
]
"""
import re
from typing import Any, Optional
from app.services.chain_templates import (
get_chain_template,
get_chain_templates,
get_chain_templates_with_config,
)
# Valid step types used across templates
CHAIN_STEP_TYPES = {
"http_request",
"webhook",
"email",
"wait",
"log",
"condition",
"parallel",
"agent_call",
"workflow",
}
# Keyword → category mapping for intent detection
_INTENT_KEYWORDS = {
"onboarding": [
"onboard",
"onboarding",
"welcome",
"signup",
"sign-up",
"new client",
"new customer",
"register",
"registration",
],
"notifications": [
"notify",
"notification",
"alert",
"team",
"inform",
"tell",
"ping",
"tell my team",
"notify team",
],
"nurture": ["follow up", "follow-up", "nurture", "nudge", "remind", "reminder", "lead", "engage", "engagement"],
}
# Keyword → template ID boosters
_TEMPLATE_BOOSTS = {
"client_onboarding": ["welcome", "onboard", "signup", "sign up", "new client", "welcome email"],
"quote_approval": ["quote", "proposal", "estimate", "approval", "approve"],
"signup_welcome": ["signup", "sign up", "welcome sequence", "onboarding sequence"],
"team_notification": ["notify team", "team alert", "team notification", "ping team", "alert team"],
"submission_alert": ["confirmation", "confirm", "received", "acknowledge", "ack"],
"lead_nurture": ["nurture", "lead nurture", "value content", "case study", "lead sequence"],
"follow_up_reminder": ["follow up", "follow-up", "reminder", "stale", "chase"],
}
# Field type → category hints
_FIELD_HINTS = {
"email": ["onboarding", "notifications", "nurture"],
"tel": ["notifications"],
"phone": ["notifications"],
}
def _score_template(template: dict, prompt: str, fields: list, form_type: str | None = None) -> float:
"""Score a template against the user's intent.
Returns confidence 0.0-1.0.
"""
if not prompt:
return 0.0
prompt_lower = prompt.lower()
template_id = template["id"]
template_name_lower = template["name"].lower()
template_desc_lower = template["description"].lower()
template_category = template.get("category", "")
score = 0.0
matches = 0
# 1. Keyword match against template boosts (strongest signal)
boosts = _TEMPLATE_BOOSTS.get(template_id, [])
for keyword in boosts:
if keyword in prompt_lower:
score += 0.3
matches += 1
# 2. Intent category match
for category, keywords in _INTENT_KEYWORDS.items():
if category == template_category:
for keyword in keywords:
if keyword in prompt_lower:
score += 0.2
matches += 1
break # One match per category is enough
# 3. Form type explicit match (user said form_type="onboarding")
if form_type:
if form_type.lower() == template_category:
score += 0.25
matches += 1
# 4. Template name/description keyword overlap with prompt
prompt_words = set(re.findall(r"\w+", prompt_lower))
name_words = set(re.findall(r"\w+", template_name_lower))
desc_words = set(re.findall(r"\w+", template_desc_lower))
overlap = prompt_words & (name_words | desc_words)
if len(name_words | desc_words) > 0:
score += 0.1 * min(len(overlap) / 3, 1.0)
# 5. Field type hints — do the form fields suggest this category?
field_types = {f.get("type", "").lower() for f in fields}
for ft, categories in _FIELD_HINTS.items():
if ft in field_types and template_category in categories:
score += 0.05
matches += 1
# Normalize to 0-1 range
# Base score from matches, capped at 1.0
if matches == 0 and score == 0:
# Check for very loose keyword overlap
for keyword in boosts:
if any(word in prompt_lower for word in keyword.split()):
score = max(score, 0.1)
break
return min(score, 1.0)
def _find_required_config(template: dict, fields: list) -> list:
"""Find template variables that need configuration.
Returns list of variable names that are not satisfied by form fields.
"""
field_names = {f.get("name", "").lower() for f in fields}
field_names.add("email") # Common implicit field
field_names.add("name")
# Extract {{variable}} from all step configs
required = set()
for step in template.get("config", {}).get("steps", []):
config = step.get("config", {})
def extract_vars(obj):
if isinstance(obj, str):
for match in re.findall(r"\{\{([^}]+)\}\}", obj):
var = match.strip().split(".")[0] # Get top-level key
if var not in ("submission", "site", "fields"):
required.add(var)
elif isinstance(obj, dict):
for v in obj.values():
extract_vars(v)
elif isinstance(obj, list):
for item in obj:
extract_vars(item)
extract_vars(config)
# Filter out fields that are already provided by the form
missing = []
for var in required:
if var not in field_names:
missing.append(var)
return sorted(missing)
def suggest_chains(
prompt: str,
fields: list,
form_type: str | None = None,
use_fallback: bool = True,
max_suggestions: int = 3,
) -> list:
"""Suggest action chains based on user intent and form fields.
Args:
prompt: Natural language description of desired workflow.
fields: List of form field definitions.
form_type: Optional form type hint (e.g. "onboarding", "marketing").
use_fallback: If True, use keyword matching (no LLM required).
max_suggestions: Maximum number of suggestions to return.
Returns:
List of suggestion dicts with name, description, steps, confidence,
requires_config, and source.
"""
# Empty prompt returns empty
if not prompt or not prompt.strip():
return []
templates = get_chain_templates_with_config()
if not templates:
return []
# Score each template
scored = []
for template in templates:
confidence = _score_template(template, prompt, fields, form_type)
if confidence > 0:
full_template = get_chain_template(template["id"])
if full_template:
requires_config = _find_required_config(full_template, fields)
scored.append(
(
confidence,
{
"name": template["name"],
"description": template["description"],
"steps": full_template["config"]["steps"],
"confidence": round(confidence, 2),
"requires_config": requires_config,
"source": "template",
},
)
)
# Sort by confidence descending
scored.sort(key=lambda x: x[0], reverse=True)
# Return top N
return [item[1] for item in scored[:max_suggestions]]
def find_template_by_category(category: str, slug: str | None = None) -> dict | None:
"""Find a template by category. If slug provided, find exact match.
Args:
category: Category name (e.g. "onboarding", "notifications", "nurture").
slug: Optional specific template ID for exact match.
Returns:
Template dict or None.
"""
templates = get_chain_templates()
for t in templates:
if t["category"] == category:
if slug and t["id"] == slug:
return t
if not slug:
return t
return None