Spaces:
Sleeping
Sleeping
| 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", | |
| } | |