Spaces:
Running
Running
| import json | |
| import os | |
| import re | |
| import sqlite3 | |
| from contextlib import contextmanager | |
| from dataclasses import dataclass | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| from typing import Any, Literal | |
| from uuid import uuid4 | |
| import numpy as np | |
| from fastapi import FastAPI, Request | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from fastapi.responses import HTMLResponse, JSONResponse, Response | |
| from pydantic import BaseModel | |
| from sentence_transformers import SentenceTransformer | |
| APP_VERSION = "1.0.8-free" | |
| MODEL_NAME = os.getenv("MODEL_NAME", "sentence-transformers/multi-qa-MiniLM-L6-cos-v1") | |
| DATABASE_PATH = Path(os.getenv("DATABASE_PATH", "/app/storage/mnkylabs.sqlite3")) | |
| MIN_SCORE = float(os.getenv("MIN_SCORE", "0.43")) | |
| AMBIGUITY_MARGIN = float(os.getenv("AMBIGUITY_MARGIN", "0.035")) | |
| ALLOWED_ORIGINS = os.getenv("ALLOWED_ORIGINS", "*") | |
| ADMIN_TOKEN = os.getenv("ADMIN_TOKEN", "") | |
| BASE_DIR = Path(__file__).parent | |
| KNOWLEDGE_PATH = BASE_DIR / "data" / "knowledge.json" | |
| FAQ_PATH = BASE_DIR / "data" / "faqs.json" | |
| class ChatRequest(BaseModel): | |
| message: str | |
| conversationId: str | None = None | |
| visitorId: str | None = None | |
| pageUrl: str | None = None | |
| locale: Literal["en", "ar", "auto"] = "auto" | |
| class LeadPayload(BaseModel): | |
| conversationId: str | |
| name: str | None = None | |
| company: str | None = None | |
| email: str | None = None | |
| phone: str | None = None | |
| projectSummary: str | None = None | |
| budgetRange: str | None = None | |
| timeline: str | None = None | |
| serviceInterest: list[str] = [] | |
| preferredContactMethod: str | None = None | |
| class Chunk: | |
| id: str | |
| kind: str | |
| title: str | |
| text: str | |
| approved_answer: str | None = None | |
| service_id: str | None = None | |
| SLANG_REPLACEMENTS: list[tuple[re.Pattern[str], str]] = [ | |
| (re.compile(r"\bu\b", re.I), "you"), | |
| (re.compile(r"\bur\b", re.I), "your"), | |
| (re.compile(r"\bidk\b", re.I), "I don't know"), | |
| (re.compile(r"\brn\b", re.I), "right now"), | |
| (re.compile(r"\basap\b", re.I), "as soon as possible"), | |
| (re.compile(r"\bwanna\b", re.I), "want to"), | |
| (re.compile(r"\bgonna\b", re.I), "going to"), | |
| (re.compile(r"\bpls\b", re.I), "please"), | |
| (re.compile(r"\bplz\b", re.I), "please"), | |
| (re.compile(r"\bai bot\b", re.I), "AI chatbot"), | |
| (re.compile(r"\bwebiste\b", re.I), "website"), | |
| (re.compile(r"\bwebstie\b", re.I), "website"), | |
| ] | |
| SERVICE_SIGNALS = { | |
| "website_development": ["website", "landing page", "site", "redesign", "online presence"], | |
| "mvp_development": ["mvp", "prototype", "first version", "startup idea", "app idea"], | |
| "ai_automation": ["automation", "automate", "workflow", "operations", "manual work", "ai in my business"], | |
| "ai_chatbot": ["chatbot", "ai chatbot", "ai bot", "assistant", "answer customers"], | |
| "custom_software": ["custom software", "custom system", "dashboard", "portal", "internal tool", "booking system", "crm", "orders", "user login"], | |
| "consultation": ["not sure", "i don't know", "consultation", "guide me", "where to start", "help me decide", "advice", "what should i do first"], | |
| } | |
| SERVICE_LABELS = { | |
| "website_development": "website development", | |
| "mvp_development": "MVP development", | |
| "ai_automation": "AI automation", | |
| "ai_chatbot": "AI chatbot", | |
| "custom_software": "custom software", | |
| "consultation": "consultation", | |
| } | |
| def utc_now() -> str: | |
| return datetime.now(timezone.utc).isoformat() | |
| def normalize_text(text: str) -> str: | |
| normalized = text.strip() | |
| for pattern, replacement in SLANG_REPLACEMENTS: | |
| normalized = pattern.sub(replacement, normalized) | |
| normalized = normalized.replace("’", "'").replace("“", '"').replace("”", '"') | |
| normalized = re.sub(r"[^\w\s.,?!'\":/@+_-]", " ", normalized, flags=re.UNICODE) | |
| normalized = re.sub(r"\s+", " ", normalized).strip() | |
| return normalized | |
| def is_arabic(text: str) -> bool: | |
| return bool(re.search(r"[\u0600-\u06ff]", text)) | |
| def load_json(path: Path) -> Any: | |
| return json.loads(path.read_text(encoding="utf-8")) | |
| def build_chunks() -> list[Chunk]: | |
| knowledge = load_json(KNOWLEDGE_PATH) | |
| faqs = load_json(FAQ_PATH) | |
| chunks: list[Chunk] = [] | |
| chunks.append( | |
| Chunk( | |
| id="company-overview", | |
| kind="company", | |
| title="Company overview", | |
| text=f"{knowledge['company']['name']}: {knowledge['company']['summary']} {knowledge['company']['positioning']}", | |
| ) | |
| ) | |
| for service in knowledge["services"]: | |
| chunks.append( | |
| Chunk( | |
| id=f"service-{service['id']}", | |
| kind="service", | |
| title=service["name"], | |
| text=f"{service['name']}. {service['description']} Good fit signals: {', '.join(service['recommendWhen'])}.", | |
| service_id=service["id"], | |
| ) | |
| ) | |
| chunks.extend( | |
| [ | |
| Chunk("pricing-notes", "pricing", "Pricing notes", " ".join(knowledge["pricingNotes"])), | |
| Chunk("process", "process", "Project process", " -> ".join(knowledge["process"])), | |
| Chunk("timelines", "timeline", "Timelines", " ".join(knowledge["timelines"])), | |
| Chunk("policies", "policy", "Policies", " ".join(knowledge["policies"])), | |
| Chunk("contact", "contact", "Contact and lead capture", json.dumps(knowledge["contact"], ensure_ascii=False)), | |
| ] | |
| ) | |
| for case_study in knowledge["caseStudies"]: | |
| chunks.append( | |
| Chunk( | |
| id=f"case-study-{case_study['id']}", | |
| kind="case_study", | |
| title=case_study["title"], | |
| text=case_study["approvedClaim"], | |
| ) | |
| ) | |
| for faq in faqs: | |
| chunks.append( | |
| Chunk( | |
| id=f"faq-{faq['id']}", | |
| kind="faq", | |
| title=faq["question"], | |
| text=" ".join([faq["question"], *faq["alternatePhrasings"], faq["approvedAnswer"]]), | |
| approved_answer=faq["approvedAnswer"], | |
| ) | |
| ) | |
| return chunks | |
| def cosine_scores(query_embedding: np.ndarray, matrix: np.ndarray) -> np.ndarray: | |
| query_norm = query_embedding / (np.linalg.norm(query_embedding) or 1) | |
| matrix_norm = matrix / np.clip(np.linalg.norm(matrix, axis=1, keepdims=True), 1e-12, None) | |
| return matrix_norm @ query_norm | |
| def recommend_services(message: str) -> list[str]: | |
| text = message.lower() | |
| matches: list[str] = [] | |
| for service, signals in SERVICE_SIGNALS.items(): | |
| if any(signal in text for signal in signals): | |
| matches.append(service) | |
| if any(term in text for term in ["booking", "payments", "payment", "checkout", "reservation", "appointments"]): | |
| if "custom_software" not in matches: | |
| matches.append("custom_software") | |
| if "ai" in text and any(term in text for term in ["stuff", "feature", "workflow", "dashboard", "business", "biz"]): | |
| if "ai_automation" not in matches: | |
| matches.append("ai_automation") | |
| if not matches and re.search(r"\b(help|build|need|want|project|business)\b", text): | |
| matches.append("consultation") | |
| return matches[:3] | |
| def extract_lead(conversation_id: str, message: str, services: list[str]) -> dict[str, Any]: | |
| email = re.search(r"[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}", message, re.I) | |
| phone = re.search(r"(?:\+?\d[\d\s().-]{7,}\d)", message) | |
| timeline = re.search(r"\b(as soon as possible|right now|this week|this month|next month|urgent|rush|\d+\s*(days|weeks|months))\b", message, re.I) | |
| budget = re.search(r"(?:budget|cost|price|around|about)\s*(?:is|:)?\s*([$\w\s.-]{2,35})", message, re.I) | |
| name = re.search(r"\b(?:my name is|name is|i'm)\s+([a-z][a-z\s]{1,35})\b", message, re.I) | |
| company = re.search(r"\b(?:company is|company name is|business is|business name is|my company is|my business is)\s+([a-z0-9][a-z0-9\s&.-]{1,45})\b", message, re.I) | |
| return { | |
| "id": str(uuid4()), | |
| "conversation_id": conversation_id, | |
| "name": name.group(1).strip() if name else None, | |
| "company": company.group(1).strip() if company else None, | |
| "email": email.group(0) if email else None, | |
| "phone": phone.group(0).strip() if phone else None, | |
| "project_summary": message if len(message) > 12 else None, | |
| "budget_range": budget.group(1).strip() if budget else None, | |
| "timeline": timeline.group(0) if timeline else None, | |
| "service_interest": json.dumps(services), | |
| "preferred_contact_method": "email" if "email" in message.lower() else "phone" if re.search(r"phone|call|whatsapp", message, re.I) else None, | |
| } | |
| def next_lead_question(lead: dict[str, Any]) -> str | None: | |
| if not json.loads(lead.get("service_interest") or "[]"): | |
| return "What type of project are you considering: website, MVP, AI automation, AI chatbot, custom software, or consultation?" | |
| if not lead.get("project_summary"): | |
| return "What business goal are you trying to achieve?" | |
| if not lead.get("budget_range"): | |
| return "Do you have a rough budget range in mind?" | |
| if not lead.get("timeline"): | |
| return "What timeline are you hoping for?" | |
| if not lead.get("email") and not lead.get("phone"): | |
| return "What is the best email or phone number for the mnkylabs team to contact you?" | |
| return None | |
| def needs_lead_capture(message: str, services: list[str]) -> bool: | |
| return bool(services) or bool(re.search(r"contact|call|email|phone|whatsapp|quote|start|project|budget|timeline|urgent|as soon as possible|my name is|name is|company is|company name is|business is|business name is", message, re.I)) | |
| def has_lead_detail(message: str) -> bool: | |
| return bool( | |
| re.search(r"[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}", message, re.I) | |
| or re.search(r"(?:\+?\d[\d\s().-]{7,}\d)", message) | |
| or re.search(r"\b(?:my name is|name is|i'm)\s+([a-z][a-z\s]{1,35})\b", message, re.I) | |
| or re.search(r"\b(?:company is|company name is|business is|business name is|my company is|my business is)\s+([a-z0-9][a-z0-9\s&.-]{1,45})\b", message, re.I) | |
| ) | |
| def fallback_answer(arabic: bool) -> str: | |
| if arabic: | |
| return "لست متأكدا من الإجابة من مصادر mnkylabs المعتمدة. أقدر أجمع بيانات التواصل وملخص المشروع ليقوم فريق mnkylabs بالمتابعة." | |
| return "I am not sure from the approved mnkylabs knowledge I have. I can collect your contact details and project summary so the mnkylabs team can follow up." | |
| def direct_guardrail_answer(message: str, arabic: bool) -> str | None: | |
| text = message.lower() | |
| if any(term in text for term in ["tech stack", "technology stack", "what stack", "which stack", "always use"]): | |
| return "The approved mnkylabs knowledge base does not list a fixed tech stack. mnkylabs chooses the build approach based on the project scope, features, integrations, and business needs." | |
| if any(term in text for term in ["revenue did you make", "revenue for clients", "client revenue", "client results", "exact client", "biggest client"]): | |
| return "Case studies are not published in the approved knowledge base yet. I should not claim specific client names, revenue, or results without approved case study information, but I can collect your details so the team can share relevant examples if available." | |
| if any(term in text for term in ["copy another website", "copy a website", "copy their website", "exactly copy"]): | |
| return "I can't confirm that mnkylabs can copy another website exactly from the approved knowledge. mnkylabs can help build a custom website based on your brand, goals, and required features." | |
| if any(term in text for term in ["refund", "refunds"]): | |
| return fallback_answer(arabic) | |
| return None | |
| def small_talk_answer(message: str, arabic: bool) -> str | None: | |
| text = re.sub(r"[^a-z0-9\u0600-\u06ff\s]", " ", message.lower()) | |
| text = re.sub(r"\s+", " ", text).strip() | |
| words = set(text.split()) | |
| greeting_words = {"hi", "hello", "hey", "yo", "salam", "مرحبا", "اهلا", "أهلا"} | |
| thanks_words = {"thanks", "thx", "ty", "شكرا", "شكرًا", "مشكور"} | |
| bye_words = {"bye", "goodbye", "later", "باي"} | |
| is_short = len(words) <= 4 | |
| is_greeting = is_short and ( | |
| bool(words & greeting_words) | |
| or text in {"good morning", "good afternoon", "good evening", "السلام عليكم"} | |
| ) | |
| is_thanks = is_short and ( | |
| bool(words & thanks_words) | |
| or text in {"thank you", "thank you so much", "appreciate it", "thanks a lot"} | |
| ) | |
| is_bye = is_short and ( | |
| bool(words & bye_words) | |
| or text in {"see you", "talk later", "that's all", "thats all", "مع السلامة"} | |
| ) and not any(term in text for term in ["pay later", "later payment"]) | |
| if is_greeting: | |
| if arabic: | |
| return "أهلا! أنا مساعد mnkylabs. أقدر أساعدك في المواقع، MVPs، الشات بوت، الأتمتة بالذكاء الاصطناعي، والبرمجيات المخصصة. ماذا تريد أن تبني؟" | |
| return "Hi! I'm the mnkylabs assistant. I can help with websites, MVPs, AI chatbots, AI automations, custom software, or a consultation. What are you looking to build?" | |
| if is_thanks: | |
| if arabic: | |
| return "على الرحب والسعة! إذا أردت، أقدر أساعدك تحدد نوع المشروع المناسب أو أجمع بياناتك ليتواصل معك فريق mnkylabs." | |
| return "You're welcome! If you want, I can help narrow down the right service or collect your details for the mnkylabs team." | |
| if is_bye: | |
| if arabic: | |
| return "شكرا لتواصلك مع mnkylabs. إذا احتجت مساعدة لاحقا، أنا موجود." | |
| return "Thanks for chatting with mnkylabs. If you need help later, I'll be here." | |
| return None | |
| def routed_faq_id(message: str) -> str | None: | |
| text = message.lower() | |
| if any(term in text for term in ["client revenue", "revenue results", "revenue did you make", "revenue for clients", "client names", "biggest client", "exact clients", "who have you worked with"]): | |
| return "faq-case_studies" | |
| if any(term in text for term in ["copy another website", "copy a website", "copy their website", "exactly copy"]): | |
| return None | |
| if any(term in text for term in ["only build websites", "only build website", "just build websites", "just websites"]): | |
| return "faq-services" | |
| if any(term in text for term in ["call me", "someone call", "talk to someone", "talk to human", "contact me", "reach me", "start project", "start a project", "begin project", "get started"]): | |
| return "faq-contact" | |
| if any(term in text for term in ["how much", "price", "pricing", "cost", "quote", "budget", "charge", "expensive", "packages", "cheap option", "pay later"]): | |
| return "faq-pricing" | |
| if any(term in text for term in ["what do you do", "what do you guys do", "what do you guys actually do", "what services", "services do you offer", "what can mnkylabs", "what kind of software", "what kind of digital products", "digital products can", "websites and ai stuff", "build websites and ai", "software and ai", "launch something online"]): | |
| return "faq-services" | |
| routes: list[tuple[str, list[str]]] = [ | |
| ("ai_chatbot", ["chatbot", "ai chatbot", "ai bot", "website assistant", "answer customers", "support bot", "sales bot", "shopping assistant"]), | |
| ("mvp_development", ["mvp", "minimum viable product", "prototype", "first version", "app idea", "startup idea"]), | |
| ("ai_automation", ["automation", "automate", "workflow", "manual work", "operations", "ai in my business"]), | |
| ("custom_software", ["custom software", "custom system", "dashboard", "portal", "internal tool", "booking system", "crm", "booking", "payments", "payment", "checkout", "manage my orders", "user login"]), | |
| ("website_development", ["website", "landing page", "site", "redesign", "online presence", "mobile", "responsive", "arabic english"]), | |
| ("timeline", ["how fast", "how long", "how soon", "as soon as possible", "asap", "right now", "this week", "this month", "timeline", "finish", "done fast", "days", "weeks", "urgent", "rush", "promise 1 day", "1 day delivery", "tomorrow"]), | |
| ("consultation", ["not sure", "i don't know", "where to start", "guide me", "consultation", "not clear", "advise me", "advice", "help me decide", "what should i build", "what should i do first", "have an idea", "talk to someone", "call me"]), | |
| ("process", ["process", "steps", "workflow", "how do we start", "what happens after"]), | |
| ("case_studies", ["case study", "case studies", "portfolio", "previous work", "client results", "examples of work"]), | |
| ("contact", ["contact", "talk to human", "talk to someone", "book a call", "call me", "reach you", "send my details", "start a project"]), | |
| ] | |
| for faq_id, terms in routes: | |
| if any(term in text for term in terms): | |
| return f"faq-{faq_id}" | |
| return None | |
| def has_timeline_intent(message: str) -> bool: | |
| text = message.lower() | |
| return any( | |
| term in text | |
| for term in ["how fast", "how long", "how soon", "as soon as possible", "asap", "right now", "this week", "this month", "timeline", "finish", "done fast", "days", "weeks", "urgent", "rush", "quick", "promise 1 day", "1 day delivery"] | |
| ) | |
| def timeline_followup(message: str) -> str: | |
| if "website" in message.lower() or "site" in message.lower(): | |
| return "What type of website do you need, and what features should it include?" | |
| return "What project type and key features should mnkylabs estimate for?" | |
| def format_answer(chunk: Chunk, services: list[str], lead_question: str | None, arabic: bool) -> str: | |
| answer = chunk.approved_answer or chunk.text | |
| if services and chunk.kind != "faq": | |
| service_text = ", ".join(service.replace("_", " ") for service in services) | |
| answer += f" Based on what you shared, the closest fit may be {service_text}." | |
| if lead_question: | |
| answer += f"\n\n{lead_question}" | |
| return answer | |
| def format_routed_answer(chunk: Chunk, message: str, services: list[str], lead_question: str | None, arabic: bool) -> str: | |
| answer = chunk.approved_answer or chunk.text | |
| if len(services) > 1 and chunk.id not in {"faq-pricing", "faq-timeline"}: | |
| service_names = ", ".join(SERVICE_LABELS.get(service, service.replace("_", " ")) for service in services) | |
| answer = f"Yes. mnkylabs can help with a combined project involving {service_names}. The team can scope the parts together so the website, software, AI, and automation pieces fit the same business goal." | |
| if chunk.id in {"faq-mvp_development", "faq-website_development", "faq-ai_chatbot", "faq-ai_automation", "faq-custom_software"} and has_timeline_intent(message): | |
| timeline_chunk = chunks_by_id.get("faq-timeline") | |
| if timeline_chunk and timeline_chunk.approved_answer: | |
| answer += f"\n\nTimeline note: {timeline_chunk.approved_answer}" | |
| if lead_question: | |
| answer += f"\n\n{lead_question}" | |
| return answer | |
| def lead_detail_acknowledgement(lead: dict[str, Any]) -> str: | |
| saved_parts: list[str] = [] | |
| if lead.get("name"): | |
| saved_parts.append("name") | |
| if lead.get("company"): | |
| saved_parts.append("company") | |
| if lead.get("email"): | |
| saved_parts.append("email") | |
| if lead.get("phone"): | |
| saved_parts.append("phone") | |
| if lead.get("budget_range"): | |
| saved_parts.append("budget range") | |
| if lead.get("timeline"): | |
| saved_parts.append("timeline") | |
| saved = ", ".join(saved_parts) if saved_parts else "details" | |
| return f"Got it, I saved your {saved}. What project are you interested in: website, MVP, AI automation, AI chatbot, custom software, or consultation?" | |
| def db() -> Any: | |
| DATABASE_PATH.parent.mkdir(parents=True, exist_ok=True) | |
| conn = sqlite3.connect(DATABASE_PATH) | |
| conn.row_factory = sqlite3.Row | |
| try: | |
| yield conn | |
| conn.commit() | |
| finally: | |
| conn.close() | |
| def init_db() -> None: | |
| with db() as conn: | |
| conn.executescript( | |
| """ | |
| create table if not exists conversations ( | |
| id text primary key, | |
| visitor_id text, | |
| page_url text, | |
| locale text, | |
| created_at text not null, | |
| updated_at text not null | |
| ); | |
| create table if not exists messages ( | |
| id integer primary key autoincrement, | |
| conversation_id text not null, | |
| role text not null, | |
| content text not null, | |
| normalized_content text, | |
| recommended_services text, | |
| sources text, | |
| created_at text not null | |
| ); | |
| create table if not exists leads ( | |
| id text primary key, | |
| conversation_id text not null, | |
| name text, | |
| company text, | |
| email text, | |
| phone text, | |
| project_summary text, | |
| budget_range text, | |
| timeline text, | |
| service_interest text, | |
| preferred_contact_method text, | |
| created_at text not null, | |
| updated_at text not null | |
| ); | |
| create table if not exists unanswered_questions ( | |
| id integer primary key autoincrement, | |
| conversation_id text, | |
| question text not null, | |
| normalized_question text not null, | |
| created_at text not null | |
| ); | |
| """ | |
| ) | |
| def store_conversation(conversation_id: str, visitor_id: str | None, page_url: str | None, locale: str) -> None: | |
| now = utc_now() | |
| with db() as conn: | |
| conn.execute( | |
| """ | |
| insert into conversations (id, visitor_id, page_url, locale, created_at, updated_at) | |
| values (?, ?, ?, ?, ?, ?) | |
| on conflict(id) do update set | |
| visitor_id = coalesce(excluded.visitor_id, conversations.visitor_id), | |
| page_url = coalesce(excluded.page_url, conversations.page_url), | |
| locale = excluded.locale, | |
| updated_at = excluded.updated_at | |
| """, | |
| (conversation_id, visitor_id, page_url, locale, now, now), | |
| ) | |
| def store_message(conversation_id: str, role: str, content: str, normalized: str | None, services: list[str], sources: list[dict[str, Any]]) -> None: | |
| with db() as conn: | |
| conn.execute( | |
| """ | |
| insert into messages (conversation_id, role, content, normalized_content, recommended_services, sources, created_at) | |
| values (?, ?, ?, ?, ?, ?, ?) | |
| """, | |
| (conversation_id, role, content, normalized, json.dumps(services), json.dumps(sources), utc_now()), | |
| ) | |
| def store_lead(lead: dict[str, Any]) -> None: | |
| now = utc_now() | |
| with db() as conn: | |
| conn.execute( | |
| """ | |
| insert into leads ( | |
| id, conversation_id, name, company, email, phone, project_summary, | |
| budget_range, timeline, service_interest, preferred_contact_method, | |
| created_at, updated_at | |
| ) | |
| values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) | |
| """, | |
| ( | |
| lead["id"], | |
| lead["conversation_id"], | |
| lead.get("name"), | |
| lead.get("company"), | |
| lead.get("email"), | |
| lead.get("phone"), | |
| lead.get("project_summary"), | |
| lead.get("budget_range"), | |
| lead.get("timeline"), | |
| lead.get("service_interest"), | |
| lead.get("preferred_contact_method"), | |
| now, | |
| now, | |
| ), | |
| ) | |
| def store_unanswered(conversation_id: str, question: str, normalized: str) -> None: | |
| with db() as conn: | |
| conn.execute( | |
| "insert into unanswered_questions (conversation_id, question, normalized_question, created_at) values (?, ?, ?, ?)", | |
| (conversation_id, question, normalized, utc_now()), | |
| ) | |
| chunks = build_chunks() | |
| chunks_by_id = {chunk.id: chunk for chunk in chunks} | |
| model = SentenceTransformer(MODEL_NAME) | |
| chunk_embeddings = model.encode([f"{chunk.title}\n{chunk.text}" for chunk in chunks], normalize_embeddings=True) | |
| init_db() | |
| app = FastAPI(title="mnkylabs Free AI Sales Assistant", version=APP_VERSION) | |
| origins = ["*"] if ALLOWED_ORIGINS == "*" else [origin.strip() for origin in ALLOWED_ORIGINS.split(",") if origin.strip()] | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=origins, | |
| allow_credentials=False, | |
| allow_methods=["GET", "POST", "OPTIONS"], | |
| allow_headers=["*"], | |
| ) | |
| def retrieve(message: str, top_k: int = 6) -> list[dict[str, Any]]: | |
| query_embedding = model.encode(message, normalize_embeddings=True) | |
| scores = cosine_scores(np.array(query_embedding), np.array(chunk_embeddings)) | |
| indices = np.argsort(scores)[::-1][:top_k] | |
| return [ | |
| { | |
| "chunk": chunks[int(index)], | |
| "score": float(scores[int(index)]), | |
| } | |
| for index in indices | |
| ] | |
| def home() -> dict[str, Any]: | |
| return { | |
| "status": "mnkylabs assistant is running", | |
| "version": APP_VERSION, | |
| "cost": "No paid AI API. Uses local sentence-transformers model.", | |
| "widget": "/widget.js", | |
| "admin": "/admin", | |
| } | |
| def chat(payload: ChatRequest) -> dict[str, Any]: | |
| conversation_id = payload.conversationId or str(uuid4()) | |
| normalized = normalize_text(payload.message) | |
| arabic = is_arabic(payload.message) | |
| guardrail_answer = direct_guardrail_answer(normalized, arabic) | |
| small_talk = small_talk_answer(normalized, arabic) | |
| services = recommend_services(normalized) | |
| results = retrieve(normalized, 6) | |
| best = results[0] | |
| second = results[1] if len(results) > 1 else None | |
| best_chunk: Chunk = best["chunk"] | |
| source_payload = [ | |
| {"id": item["chunk"].id, "kind": item["chunk"].kind, "title": item["chunk"].title, "score": round(item["score"], 4)} | |
| for item in results | |
| ] | |
| route_id = routed_faq_id(normalized) | |
| routed_chunk = chunks_by_id.get(route_id) if route_id else None | |
| store_conversation(conversation_id, payload.visitorId, payload.pageUrl, payload.locale) | |
| store_message(conversation_id, "user", payload.message, normalized, services, source_payload) | |
| lead = extract_lead(conversation_id, normalized, services) | |
| lead_detail_only = has_lead_detail(normalized) and not services | |
| capture_lead = needs_lead_capture(normalized, services) or lead_detail_only | |
| lead_question = next_lead_question(lead) if capture_lead else None | |
| ambiguous = second is not None and best["score"] < 0.62 and (best["score"] - second["score"]) < AMBIGUITY_MARGIN | |
| if guardrail_answer: | |
| answer = guardrail_answer | |
| elif lead_detail_only: | |
| answer = lead_detail_acknowledgement(lead) | |
| elif small_talk: | |
| answer = small_talk | |
| elif routed_chunk: | |
| if routed_chunk.id == "faq-services": | |
| routed_followup = None | |
| elif routed_chunk.id == "faq-timeline" and capture_lead: | |
| routed_followup = timeline_followup(normalized) | |
| else: | |
| routed_followup = lead_question | |
| answer = format_routed_answer(routed_chunk, normalized, services, routed_followup, arabic) | |
| elif best["score"] < MIN_SCORE or ambiguous: | |
| answer = fallback_answer(arabic) | |
| if lead_question: | |
| answer += f"\n\n{lead_question}" | |
| store_unanswered(conversation_id, payload.message, normalized) | |
| else: | |
| answer = format_answer(best_chunk, services, lead_question, arabic) | |
| if capture_lead: | |
| store_lead(lead) | |
| store_message(conversation_id, "assistant", answer, None, services, source_payload) | |
| return { | |
| "conversationId": conversation_id, | |
| "answer": answer, | |
| "recommendedServices": services, | |
| "needsContact": capture_lead, | |
| "lead": { | |
| "email": lead.get("email"), | |
| "phone": lead.get("phone"), | |
| "budgetRange": lead.get("budget_range"), | |
| "timeline": lead.get("timeline"), | |
| "serviceInterest": json.loads(lead.get("service_interest") or "[]"), | |
| }, | |
| "sources": source_payload, | |
| } | |
| def lead(payload: LeadPayload) -> dict[str, Any]: | |
| data = { | |
| "id": str(uuid4()), | |
| "conversation_id": payload.conversationId, | |
| "name": payload.name, | |
| "company": payload.company, | |
| "email": payload.email, | |
| "phone": payload.phone, | |
| "project_summary": payload.projectSummary, | |
| "budget_range": payload.budgetRange, | |
| "timeline": payload.timeline, | |
| "service_interest": json.dumps(payload.serviceInterest), | |
| "preferred_contact_method": payload.preferredContactMethod, | |
| } | |
| store_lead(data) | |
| return {"ok": True, "leadId": data["id"]} | |
| def analytics_data() -> dict[str, Any]: | |
| with db() as conn: | |
| leads = [dict(row) for row in conn.execute("select * from leads order by created_at desc limit 100")] | |
| conversations = [dict(row) for row in conn.execute("select * from conversations order by created_at desc limit 100")] | |
| unanswered = [dict(row) for row in conn.execute("select * from unanswered_questions order by created_at desc limit 50")] | |
| common = [ | |
| dict(row) | |
| for row in conn.execute( | |
| """ | |
| select normalized_content as question, count(*) as count | |
| from messages | |
| where role = 'user' and normalized_content is not null | |
| group by normalized_content | |
| order by count desc | |
| limit 25 | |
| """ | |
| ) | |
| ] | |
| messages_count = conn.execute("select count(*) as count from messages").fetchone()["count"] | |
| service_counts: dict[str, int] = {} | |
| with db() as conn: | |
| for row in conn.execute("select recommended_services from messages where recommended_services is not null"): | |
| for service in json.loads(row["recommended_services"] or "[]"): | |
| service_counts[service] = service_counts.get(service, 0) + 1 | |
| return { | |
| "totalLeads": len(leads), | |
| "totalConversations": len(conversations), | |
| "totalMessages": messages_count, | |
| "leads": leads, | |
| "conversations": conversations, | |
| "commonQuestions": common, | |
| "unansweredQuestions": unanswered, | |
| "recommendedServices": [{"service": key, "count": value} for key, value in sorted(service_counts.items(), key=lambda item: item[1], reverse=True)], | |
| } | |
| def analytics(request: Request) -> JSONResponse: | |
| if ADMIN_TOKEN and request.query_params.get("token") != ADMIN_TOKEN: | |
| return JSONResponse({"error": "Unauthorized"}, status_code=401) | |
| return JSONResponse(analytics_data()) | |
| def admin() -> str: | |
| data = analytics_data() | |
| lead_rows = "".join( | |
| f"<tr><td>{lead.get('created_at','')}</td><td>{lead.get('name') or ''}</td><td>{lead.get('email') or lead.get('phone') or ''}</td><td>{lead.get('project_summary') or ''}</td></tr>" | |
| for lead in data["leads"][:30] | |
| ) | |
| unanswered_rows = "".join( | |
| f"<tr><td>{item.get('created_at','')}</td><td>{item.get('question','')}</td></tr>" | |
| for item in data["unansweredQuestions"][:30] | |
| ) | |
| return f""" | |
| <!doctype html> | |
| <html> | |
| <head> | |
| <title>mnkylabs Assistant Admin</title> | |
| <meta name="viewport" content="width=device-width, initial-scale=1" /> | |
| <style> | |
| body {{ margin:0; font-family: Inter, system-ui, sans-serif; background:#0c0f14; color:#eef2f7; }} | |
| main {{ max-width:1120px; margin:auto; padding:28px; }} | |
| .grid {{ display:grid; grid-template-columns:repeat(4,1fr); gap:12px; }} | |
| .card {{ background:#151b25; border:1px solid #283244; border-radius:8px; padding:16px; }} | |
| .card strong {{ display:block; font-size:30px; margin-top:8px; }} | |
| table {{ width:100%; border-collapse:collapse; margin-top:10px; }} | |
| td, th {{ padding:10px; border-bottom:1px solid #283244; text-align:left; vertical-align:top; }} | |
| th, .muted {{ color:#9ca8b8; }} | |
| @media(max-width:760px) {{ .grid {{ grid-template-columns:1fr 1fr; }} }} | |
| </style> | |
| </head> | |
| <body> | |
| <main> | |
| <h1>mnkylabs Assistant Admin</h1> | |
| <p class="muted">Free Hugging Face version. SQLite storage may reset if the Space rebuilds unless persistent storage is configured.</p> | |
| <section class="grid"> | |
| <div class="card">Leads<strong>{data["totalLeads"]}</strong></div> | |
| <div class="card">Conversations<strong>{data["totalConversations"]}</strong></div> | |
| <div class="card">Messages<strong>{data["totalMessages"]}</strong></div> | |
| <div class="card">Unanswered<strong>{len(data["unansweredQuestions"])}</strong></div> | |
| </section> | |
| <section class="card" style="margin-top:16px"> | |
| <h2>Recent leads</h2> | |
| <table><thead><tr><th>Time</th><th>Name</th><th>Contact</th><th>Summary</th></tr></thead><tbody>{lead_rows}</tbody></table> | |
| </section> | |
| <section class="card" style="margin-top:16px"> | |
| <h2>Unanswered questions</h2> | |
| <table><thead><tr><th>Time</th><th>Question</th></tr></thead><tbody>{unanswered_rows}</tbody></table> | |
| </section> | |
| </main> | |
| </body> | |
| </html> | |
| """ | |
| def widget_js() -> Response: | |
| script = r""" | |
| (function () { | |
| if (window.MnkylabsAssistantLoaded) return; | |
| window.MnkylabsAssistantLoaded = true; | |
| var script = document.currentScript; | |
| var apiBase = new URL(script.src).origin; | |
| var color = script.getAttribute("data-color") || "#68e0cf"; | |
| var title = script.getAttribute("data-title") || "mnkylabs Assistant"; | |
| var welcome = script.getAttribute("data-welcome") || "Hi, I'm the mnkylabs assistant. Tell me what you want to build, automate, or launch."; | |
| var conversationId = localStorage.getItem("mnkylabs_conversation_id") || ""; | |
| var style = document.createElement("style"); | |
| style.textContent = ` | |
| .mnky-chat-button{position:fixed;right:20px;bottom:20px;z-index:2147483000;width:58px;height:58px;border-radius:50%;border:0;background:${color};color:#07100f;font-weight:800;box-shadow:0 14px 40px rgba(0,0,0,.28);cursor:pointer} | |
| .mnky-chat-panel{position:fixed;right:20px;bottom:90px;z-index:2147483000;width:min(380px,calc(100vw - 32px));height:min(620px,calc(100vh - 120px));display:none;flex-direction:column;background:#0c0f14;color:#eef2f7;border:1px solid #283244;border-radius:8px;overflow:hidden;box-shadow:0 18px 60px rgba(0,0,0,.38);font-family:Inter,system-ui,sans-serif} | |
| .mnky-chat-header{display:flex;align-items:center;justify-content:space-between;padding:14px 16px;background:#151b25;border-bottom:1px solid #283244} | |
| .mnky-chat-header strong{font-size:15px}.mnky-chat-close{background:transparent;border:0;color:#9ca8b8;font-size:22px;cursor:pointer} | |
| .mnky-chat-messages{flex:1;overflow:auto;padding:14px;display:flex;flex-direction:column;gap:10px} | |
| .mnky-msg{max-width:82%;padding:10px 12px;border-radius:8px;font-size:14px;line-height:1.45;white-space:pre-wrap} | |
| .mnky-user{align-self:flex-end;background:${color};color:#07100f}.mnky-bot{align-self:flex-start;background:#1b2230;color:#eef2f7} | |
| .mnky-chat-form{display:flex;gap:8px;padding:12px;border-top:1px solid #283244;background:#151b25} | |
| .mnky-chat-input{flex:1;min-width:0;border:1px solid #283244;border-radius:8px;background:#0c0f14;color:#eef2f7;padding:10px} | |
| .mnky-chat-send{border:0;border-radius:8px;background:${color};color:#07100f;font-weight:700;padding:0 14px;cursor:pointer} | |
| `; | |
| document.head.appendChild(style); | |
| var button = document.createElement("button"); | |
| button.className = "mnky-chat-button"; | |
| button.type = "button"; | |
| button.textContent = "AI"; | |
| button.setAttribute("aria-label", "Open mnkylabs chat"); | |
| var panel = document.createElement("section"); | |
| panel.className = "mnky-chat-panel"; | |
| panel.innerHTML = ` | |
| <div class="mnky-chat-header"><strong>${title}</strong><button class="mnky-chat-close" type="button" aria-label="Close">x</button></div> | |
| <div class="mnky-chat-messages"></div> | |
| <form class="mnky-chat-form"> | |
| <input class="mnky-chat-input" placeholder="Ask about websites, MVPs, AI, automation..." /> | |
| <button class="mnky-chat-send" type="submit">Send</button> | |
| </form> | |
| `; | |
| document.body.appendChild(button); | |
| document.body.appendChild(panel); | |
| var messages = panel.querySelector(".mnky-chat-messages"); | |
| var form = panel.querySelector(".mnky-chat-form"); | |
| var input = panel.querySelector(".mnky-chat-input"); | |
| var close = panel.querySelector(".mnky-chat-close"); | |
| function addMessage(text, role) { | |
| var item = document.createElement("div"); | |
| item.className = "mnky-msg " + (role === "user" ? "mnky-user" : "mnky-bot"); | |
| item.textContent = text; | |
| messages.appendChild(item); | |
| messages.scrollTop = messages.scrollHeight; | |
| } | |
| addMessage(welcome, "bot"); | |
| button.addEventListener("click", function () { | |
| panel.style.display = panel.style.display === "flex" ? "none" : "flex"; | |
| input.focus(); | |
| }); | |
| close.addEventListener("click", function () { panel.style.display = "none"; }); | |
| form.addEventListener("submit", async function (event) { | |
| event.preventDefault(); | |
| var text = input.value.trim(); | |
| if (!text) return; | |
| input.value = ""; | |
| addMessage(text, "user"); | |
| addMessage("Thinking...", "bot"); | |
| var loading = messages.lastChild; | |
| try { | |
| var res = await fetch(apiBase + "/chat", { | |
| method: "POST", | |
| headers: { "Content-Type": "application/json" }, | |
| body: JSON.stringify({ message: text, conversationId: conversationId, pageUrl: location.href, locale: "auto" }) | |
| }); | |
| var data = await res.json(); | |
| conversationId = data.conversationId; | |
| localStorage.setItem("mnkylabs_conversation_id", conversationId); | |
| loading.textContent = data.answer || "I am not sure. Please share your contact details and the mnkylabs team can follow up."; | |
| } catch (error) { | |
| loading.textContent = "Sorry, I could not connect right now. Please try again in a moment."; | |
| } | |
| }); | |
| })(); | |
| """ | |
| return Response(content=script, media_type="application/javascript") | |