File size: 6,106 Bytes
fbd808d 76398fd fbd808d 76398fd fbd808d 76398fd fbd808d 76398fd fbd808d 76398fd 311871d 76398fd 311871d 76398fd fbd808d 76398fd fbd808d 76398fd fbd808d 76398fd fbd808d 76398fd fbd808d 76398fd fbd808d 76398fd fbd808d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 | """
services/tool_executor.py
──────────────────────────
Tool dispatch layer — executes tool calls and normalizes assistant messages.
"""
import asyncio
import json
import re
import urllib.parse
from services.observability import setup_logger
from services.supabase_service import execute_tool_rpc
from services.tool_compressor import compress_tool_output, geocode_address_handler
from config import CONTACT_EMAIL, CONTACT_WHATSAPP
logger = setup_logger(__name__)
def normalize_assistant_message(msg) -> dict:
"""Converts a ChatCompletionMessage SDK object to a clean dict, stripping internal fields."""
if getattr(msg, "tool_calls", None):
return {
"role": "assistant",
"content": msg.content or "",
"tool_calls": [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments,
},
}
for tc in msg.tool_calls
],
}
return {
"role": "assistant",
"content": getattr(msg, "content", "") or "",
}
def parse_tool_args(raw_arguments) -> dict:
"""Safely parse tool call arguments — handles both str and dict."""
if isinstance(raw_arguments, str):
try:
return json.loads(raw_arguments)
except Exception:
return {}
return raw_arguments or {}
def _suggest_actions_handler(func_args: dict) -> dict:
"""Handles suggest_actions tool by registering action chips."""
actions = func_args.get("actions") or func_args.get("options") or []
return {
"status": "success",
"message": (
"Interactive action button chips have been registered and will be rendered automatically "
"by the UI below the message. Do NOT write markdown links, dummy URLs, or repetitive link lists in your text response."
),
"actions": actions,
}
def _contact_buttons_handler(func_args: dict) -> dict:
"""Generates markdown contact buttons for email and WhatsApp."""
raw_message = func_args.get("message", "Hi, I'd like to discuss a custom build.")
encoded_message = urllib.parse.quote(raw_message)
return {
"status": "success",
"email_button_markdown": f"[Get in touch via Email](mailto:{CONTACT_EMAIL})",
"whatsapp_button_markdown": f"[Chat on WhatsApp](https://wa.me/{CONTACT_WHATSAPP}?text={encoded_message})",
}
def _geocode_handler(func_args: dict) -> dict:
"""Forwards geocode requests to the geocode_address_handler."""
return geocode_address_handler(func_args.get("address", ""))
async def _data_export_handler(func_args: dict) -> dict:
"""Uploads export markdown to Appwrite storage."""
from services.appwrite_service import upload_document_to_appwrite
content = func_args.get("content", "")
return await upload_document_to_appwrite(content, "md")
async def _alert_subscription_handler(func_args: dict) -> dict:
"""Forwards alert subscription creation to alert_service."""
from services.alert_service import create_pending_subscription
return await create_pending_subscription(
email=func_args.get("email", ""),
criteria_type=func_args.get("criteria_type", ""),
criteria=func_args.get("criteria", {}),
raw_text=func_args.get("raw_request_text", ""),
)
_LOCAL_HANDLERS = {
"suggest_actions": _suggest_actions_handler,
"generate_contact_buttons": _contact_buttons_handler,
"geocode_address": _geocode_handler,
"generate_data_export": _data_export_handler,
"create_alert_subscription": _alert_subscription_handler,
}
async def execute_tool_by_name(func_name: str, func_args: dict) -> dict:
"""Dispatches a tool call by name, routing to local handlers or Supabase RPC."""
handler = _LOCAL_HANDLERS.get(func_name)
if handler:
if asyncio.iscoroutinefunction(handler):
return await handler(func_args)
return handler(func_args)
return await execute_tool_rpc(func_name, func_args)
def _parse_json_direct(failed_gen: str) -> tuple[str | None, dict | None]:
"""Attempts direct JSON parse of failed generation payload."""
try:
data = json.loads(failed_gen)
if isinstance(data, dict) and "name" in data:
args = data.get("arguments", {})
if isinstance(args, str):
try:
args = json.loads(args)
except Exception:
pass
return data["name"], args if isinstance(args, dict) else {}
except Exception:
pass
return None, None
def _parse_json_regex(failed_gen: str) -> tuple[str | None, dict | None]:
"""Attempts regex extraction of JSON name/arguments dictionary."""
pattern = r'\{\s*"name"\s*:\s*"([a-zA-Z0-9_]+)"\s*,\s*"arguments"\s*:\s*(\{.*?\})\s*\}'
match = re.search(pattern, failed_gen, re.DOTALL)
if match:
try:
return match.group(1), json.loads(match.group(2))
except Exception:
pass
return None, None
def _parse_function_tag(failed_gen: str) -> tuple[str | None, dict | None]:
"""Attempts regex extraction of legacy <function=name>{args} tag format."""
pattern = r"<function=([a-zA-Z0-9_]+)[>\s]*(\{.*?\})"
match = re.search(pattern, failed_gen, re.DOTALL)
if match:
try:
return match.group(1), json.loads(match.group(2))
except Exception:
pass
return None, None
def parse_failed_generation(failed_gen: str) -> tuple[str | None, dict | None]:
"""Extracts tool name and arguments from a Groq 400 failed_generation string."""
if not failed_gen:
return None, None
for parser in (_parse_json_direct, _parse_json_regex, _parse_function_tag):
name, args = parser(failed_gen)
if name is not None:
return name, args
return None, None
|