leocrm-bot
231 lines
8.1 KiB
Python
231 lines
8.1 KiB
Python
"""Action mapper — maps natural language intents to proposed API calls.
|
|
|
|
Used by the mock LLM client for test mode and as a fallback.
|
|
Supports keyword-based intent detection for common CRM operations.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any
|
|
|
|
# Precompiled patterns for intent detection
|
|
_PATTERNS = {
|
|
"create_company": re.compile(
|
|
r"\b(create|add|new)\b.*\b(company|firm|organization|organisation)\b", re.IGNORECASE
|
|
),
|
|
"delete_company": re.compile(r"\b(delete|remove)\b.*\b(company|firm)\b", re.IGNORECASE),
|
|
"update_company": re.compile(
|
|
r"\b(update|edit|modify|change)\b.*\b(company|firm)\b", re.IGNORECASE
|
|
),
|
|
"list_company": re.compile(
|
|
r"\b(list|show|find|search|get|display)\b.*\b(compan|firm)\b", re.IGNORECASE
|
|
),
|
|
"list_company2": re.compile(r"\bcompan.*\b(list|all)\b", re.IGNORECASE),
|
|
"create_contact": re.compile(r"\b(create|add|new)\b.*\b(contact|person)\b", re.IGNORECASE),
|
|
"list_contact": re.compile(
|
|
r"\b(list|show|find|search|get|display)\b.*\b(contact|person)\b", re.IGNORECASE
|
|
),
|
|
"list_workflow": re.compile(r"\b(list|show|get|display)\b.*\b(workflow)\b", re.IGNORECASE),
|
|
"create_workflow": re.compile(r"\b(create|new|add)\b.*\b(workflow)\b", re.IGNORECASE),
|
|
"help": re.compile(r"\b(help|what can you do|assist)\b", re.IGNORECASE),
|
|
}
|
|
|
|
# Name extraction patterns - using single-quoted strings to avoid escaping issues
|
|
_NAME_PATTERNS = [
|
|
re.compile(r"\b(?:named|called|for)\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE),
|
|
re.compile(r"\bcompany\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE),
|
|
re.compile(r"\bcontact\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE),
|
|
re.compile(r"\bworkflow\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE),
|
|
re.compile(r"\bfirm\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE),
|
|
re.compile(r"\bperson\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE),
|
|
]
|
|
|
|
# Field extraction patterns
|
|
_INDUSTRY_PAT = re.compile(r"industry\s+(?:to|:)?\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE)
|
|
_NAME_UPDATE_PAT = re.compile(
|
|
r"(?:name|rename)\s+(?:to|:)?\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE
|
|
)
|
|
_PHONE_PAT = re.compile(r"phone\s+(?:to|:)?\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE)
|
|
_EMAIL_PAT = re.compile(r"email\s+(?:to|:)?\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE)
|
|
_SEARCH_PAT = re.compile(
|
|
r"\b(?:named|called|matching|with name)\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE
|
|
)
|
|
|
|
|
|
def map_query_to_actions(query: str, context: dict[str, Any] | None = None) -> list[dict[str, Any]]:
|
|
"""Map a natural language query to proposed API actions.
|
|
|
|
Uses keyword matching to detect intents. Returns a list of proposed
|
|
action dictionaries with method, path, body, description, and confidence.
|
|
"""
|
|
context = context or {}
|
|
q = query.lower().strip()
|
|
actions: list[dict[str, Any]] = []
|
|
|
|
# --- Company intents ---
|
|
if _PATTERNS["create_company"].search(q):
|
|
name = _extract_name(query)
|
|
actions.append(
|
|
{
|
|
"method": "POST",
|
|
"path": "/api/v1/companies",
|
|
"body": {"name": name or "New Company"},
|
|
"description": f"Create a new company named '{name or 'New Company'}'",
|
|
"confidence": 0.9,
|
|
}
|
|
)
|
|
|
|
elif _PATTERNS["delete_company"].search(q):
|
|
entity_id = context.get("company_id") or context.get("entity_id")
|
|
if entity_id:
|
|
actions.append(
|
|
{
|
|
"method": "DELETE",
|
|
"path": f"/api/v1/companies/{entity_id}",
|
|
"body": None,
|
|
"description": f"Delete company {entity_id}",
|
|
"confidence": 0.9,
|
|
}
|
|
)
|
|
else:
|
|
actions.append(
|
|
{
|
|
"method": "DELETE",
|
|
"path": "/api/v1/companies/{id}",
|
|
"body": None,
|
|
"description": "Delete a company (requires company ID in context or selection)",
|
|
"confidence": 0.5,
|
|
}
|
|
)
|
|
|
|
elif _PATTERNS["update_company"].search(q):
|
|
entity_id = context.get("company_id") or context.get("entity_id")
|
|
path = f"/api/v1/companies/{entity_id}" if entity_id else "/api/v1/companies/{id}"
|
|
actions.append(
|
|
{
|
|
"method": "PATCH",
|
|
"path": path,
|
|
"body": _extract_update_fields(query),
|
|
"description": "Update company information",
|
|
"confidence": 0.8,
|
|
}
|
|
)
|
|
|
|
elif _PATTERNS["list_company"].search(q) or _PATTERNS["list_company2"].search(q):
|
|
search_term = _extract_search_term(query)
|
|
desc = "List companies"
|
|
if search_term:
|
|
desc += f" matching '{search_term}'"
|
|
actions.append(
|
|
{
|
|
"method": "GET",
|
|
"path": "/api/v1/companies",
|
|
"body": None,
|
|
"description": desc,
|
|
"confidence": 0.85,
|
|
}
|
|
)
|
|
|
|
# --- Contact intents ---
|
|
elif _PATTERNS["create_contact"].search(q):
|
|
name = _extract_name(query)
|
|
actions.append(
|
|
{
|
|
"method": "POST",
|
|
"path": "/api/v1/contacts",
|
|
"body": {"name": name or "New Contact"},
|
|
"description": f"Create a new contact named '{name or 'New Contact'}'",
|
|
"confidence": 0.9,
|
|
}
|
|
)
|
|
|
|
elif _PATTERNS["list_contact"].search(q):
|
|
actions.append(
|
|
{
|
|
"method": "GET",
|
|
"path": "/api/v1/contacts",
|
|
"body": None,
|
|
"description": "List contacts",
|
|
"confidence": 0.85,
|
|
}
|
|
)
|
|
|
|
# --- Workflow intents ---
|
|
elif _PATTERNS["list_workflow"].search(q):
|
|
actions.append(
|
|
{
|
|
"method": "GET",
|
|
"path": "/api/v1/workflows",
|
|
"body": None,
|
|
"description": "List workflows",
|
|
"confidence": 0.85,
|
|
}
|
|
)
|
|
|
|
elif _PATTERNS["create_workflow"].search(q):
|
|
name = _extract_name(query)
|
|
actions.append(
|
|
{
|
|
"method": "POST",
|
|
"path": "/api/v1/workflows",
|
|
"body": {"name": name or "New Workflow", "steps": []},
|
|
"description": "Create a new workflow",
|
|
"confidence": 0.8,
|
|
}
|
|
)
|
|
|
|
# --- Generic fallback ---
|
|
if not actions:
|
|
if _PATTERNS["help"].search(q):
|
|
actions.append(
|
|
{
|
|
"method": "GET",
|
|
"path": "/api/v1/companies",
|
|
"body": None,
|
|
"description": "Show available companies (demonstration action)",
|
|
"confidence": 0.3,
|
|
}
|
|
)
|
|
|
|
return actions
|
|
|
|
|
|
def _extract_name(query: str) -> str | None:
|
|
"""Extract a name from the query, looking for 'named X', 'called X', 'for X'."""
|
|
for pat in _NAME_PATTERNS:
|
|
match = pat.search(query)
|
|
if match:
|
|
return match.group(1).strip()
|
|
return None
|
|
|
|
|
|
def _extract_search_term(query: str) -> str | None:
|
|
"""Extract a search term from the query."""
|
|
match = _SEARCH_PAT.search(query)
|
|
if match:
|
|
return match.group(1).strip()
|
|
return None
|
|
|
|
|
|
def _extract_update_fields(query: str) -> dict[str, Any]:
|
|
"""Extract fields to update from the query."""
|
|
fields: dict[str, Any] = {}
|
|
if re.search(r"\bindustry\b", query, re.IGNORECASE):
|
|
match = _INDUSTRY_PAT.search(query)
|
|
if match:
|
|
fields["industry"] = match.group(1).strip()
|
|
if re.search(r"\b(name|rename)\b", query, re.IGNORECASE):
|
|
match = _NAME_UPDATE_PAT.search(query)
|
|
if match:
|
|
fields["name"] = match.group(1).strip()
|
|
if re.search(r"\bphone\b", query, re.IGNORECASE):
|
|
match = _PHONE_PAT.search(query)
|
|
if match:
|
|
fields["phone"] = match.group(1).strip()
|
|
if re.search(r"\bemail\b", query, re.IGNORECASE):
|
|
match = _EMAIL_PAT.search(query)
|
|
if match:
|
|
fields["email"] = match.group(1).strip()
|
|
return fields or {"name": "Updated Name"}
|