Spaces:
Sleeping
Sleeping
File size: 4,380 Bytes
83fe4f9 fb38df2 83fe4f9 fb38df2 83fe4f9 fb38df2 83fe4f9 | 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 | from __future__ import annotations
from backend.ai.client import llm_client, render_prompt
from backend.config import get_settings
from backend.ai.prompts import ANALYZE_PROMPT, CUSTOM_TASK_PROMPT
from backend.models.schemas import TaskCategory
def analyze_business_description(description: str) -> dict:
settings = get_settings()
if settings.analyze_with_ai and llm_client.is_ready():
try:
return llm_client.generate_json(render_prompt(ANALYZE_PROMPT, description=description))
except Exception:
pass
lower = description.lower()
fully_automatable = []
ai_assisted = []
manual = []
if "order" in lower:
fully_automatable.append(
{
"id": "task_order_processing",
"name": "Order processing",
"description": "Read order emails, check stock, log rows, and reply.",
"time_estimate": "10-15 hours/week",
"pain_level": "high",
}
)
if "friday" in lower or "weekly" in lower:
fully_automatable.append(
{
"id": "task_weekly_summary",
"name": "Weekly order summary report",
"description": "Compile the week's orders and send a summary.",
"time_estimate": "1-2 hours/week",
"pain_level": "medium",
}
)
ai_assisted.append(
{
"id": "task_availability_inquiries",
"name": "Availability inquiries",
"description": "Draft replies using current stock and pricing.",
"time_estimate": "2-3 hours/week",
"pain_level": "medium",
}
)
ai_assisted.append(
{
"id": "task_low_inventory_alerts",
"name": "Low inventory alerts",
"description": "Watch for thin inventory and propose next steps.",
"time_estimate": "30 minutes/week",
"pain_level": "medium",
}
)
manual.append(
{
"id": "task_special_requests",
"name": "Custom and relationship-sensitive requests",
"description": "Gift boxes, delivery arrangements, and VIP relationship notes.",
"reason_manual": "These need human judgment, pricing discretion, and a personal touch.",
}
)
return {
"business_type": "Email-driven small business",
"summary": "FlowPilot found repetitive inbox work that maps well to Gmail, Sheets, and scheduled automations.",
"tasks": {
"fully_automatable": fully_automatable,
"ai_assisted": ai_assisted,
"manual": manual,
},
}
def analyze_custom_task(business_description: str, existing_workflows: list[dict], custom_task: str) -> dict:
if llm_client.is_ready():
return llm_client.generate_json(
render_prompt(
CUSTOM_TASK_PROMPT,
business_description=business_description,
existing_workflows=existing_workflows,
primitives_list=[
"email_received",
"schedule",
"file_uploaded",
"manual_trigger",
"ai_extract",
"ai_classify",
"ai_compose",
"sheets_read",
"sheets_write",
"sheets_update",
"send_email",
"notify_owner",
"condition",
"loop",
"wait_for_input",
],
custom_task=custom_task,
)
)
lower = custom_task.lower()
feasible = any(word in lower for word in ("flag", "priority", "sheet", "email"))
category = TaskCategory.FULLY_AUTOMATABLE if feasible else TaskCategory.MANUAL
return {
"feasible": feasible,
"category": category,
"task_name": "Restaurant priority handling",
"task_description": custom_task,
"missing_capabilities": [] if feasible else ["A matching primitive for this request is missing."],
"reason_if_manual": None if feasible else "This task needs a capability outside the current primitives.",
"time_estimate": "1 hour/week",
"pain_level": "medium",
}
|