Spaces:
Running
Running
| import os | |
| import json | |
| import logging | |
| import asyncio | |
| import time | |
| import re | |
| from typing import List, Dict, Any, Optional | |
| from fastapi import FastAPI, Request | |
| from pydantic import BaseModel | |
| import httpx | |
| from dotenv import load_dotenv | |
| load_dotenv() | |
| # Vera Engine - Platinum Cloud Deployment Build | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger("VeraEngine") | |
| app = FastAPI() | |
| class ContextPayload(BaseModel): | |
| scope: str | |
| context_id: str | |
| version: int | |
| payload: Dict[Any, Any] | |
| class TickRequest(BaseModel): | |
| available_triggers: List[str] | |
| # API Keys from Environment | |
| OPENROUTER_KEYS = [os.getenv(f"OPENROUTER_KEY_{i}") for i in range(1, 7)] | |
| OPENROUTER_KEYS = [k for k in OPENROUTER_KEYS if k] or [os.getenv("OPENROUTER_API_KEY")] | |
| OPENROUTER_KEYS = [k for k in OPENROUTER_KEYS if k] | |
| class KeyRotator: | |
| def __init__(self, keys: List[str]): | |
| self.keys, self.index, self.lock = keys, 0, asyncio.Lock() | |
| async def get_key(self) -> str: | |
| if not self.keys: return "" | |
| async with self.lock: | |
| key = self.keys[self.index] | |
| self.index = (self.index + 1) % len(self.keys) | |
| return key | |
| or_rotator = KeyRotator(OPENROUTER_KEYS) | |
| contexts = {"category": {}, "merchant": {}, "customer": {}, "trigger": {}} | |
| llm_semaphore = asyncio.Semaphore(1) | |
| PRIORITY_MAP = { | |
| "drop_in_orders": 100, "revenue_drop": 95, "regulation_change": 90, | |
| "recall_due": 80, "high_churn": 70, "competitor_action": 65, "perf_spike": 60 | |
| } | |
| async def call_llm(prompt: str, system: str = "") -> Optional[Dict]: | |
| async with llm_semaphore: | |
| key = await or_rotator.get_key() | |
| if not key: return None | |
| headers = {"Authorization": f"Bearer {key}", "Content-Type": "application/json", "HTTP-Referer": "https://vera.ai", "X-Title": "Vera Engine"} | |
| payload = { | |
| "model": "google/gemini-2.0-flash-exp:free", | |
| "messages": [ | |
| {"role": "system", "content": system + "\n\nRETURN JSON ONLY. NO MARKDOWN. NO FABRICATION."}, | |
| {"role": "user", "content": prompt} | |
| ], | |
| "temperature": 0.0, | |
| "response_format": {"type": "json_object"} | |
| } | |
| async with httpx.AsyncClient() as client: | |
| try: | |
| response = await client.post("https://openrouter.ai/api/v1/chat/completions", headers=headers, json=payload, timeout=60.0) | |
| if response.status_code == 200: | |
| return json.loads(response.json()["choices"][0]["message"]["content"]) | |
| return None | |
| except Exception: return None | |
| def prune_message(text: str) -> str: | |
| text = text.replace("\n", " ").strip() | |
| if len(text) <= 315: return text | |
| match = re.findall(r'.*?[.!?]', text) | |
| res = "" | |
| for m in match: | |
| if len(res) + len(m) <= 312: res += m | |
| else: break | |
| return res if res else text[:309] + "..." | |
| async def compose(category: dict, merchant: dict, trigger: dict, customer: Optional[dict] = None) -> dict: | |
| ident, perf = merchant.get("identity", {}), merchant.get("performance", {}) | |
| owner_name = ident.get("owner_first_name", "Partner") | |
| m = {k: v for k, v in perf.items() if v and v != "0" and v != 0 and v != "?"} | |
| payload = trigger.get("payload", {}) | |
| prefix = "Dr." if category.get('slug') == 'dentists' else "" | |
| lang_pref = ident.get('languages', ['en']) | |
| hinglish_note = "USE NATURAL HINGLISH (50% English words, 50% Hindi script/transliteration)." if 'hi' in lang_pref else "USE PROFESSIONAL ENGLISH." | |
| sys_prompt = f"""Role: Elite Growth Strategist. Target: {prefix} {owner_name}. | |
| RULES: | |
| 1. NO FABRICATION: Use metrics provided: {json.dumps(m)}. | |
| 2. CITATION: Include verbatim source from {json.dumps(payload)}. | |
| 3. WHY ACT: Link {trigger.get('kind')} to revenue loss or business growth. | |
| 4. MERCHANT FIT: Mention locality {ident.get('locality')}. {hinglish_note} | |
| 5. LENGTH: 280-310 chars. | |
| JSON: {{"body": "...", "cta": "...", "rationale": "..."}}""" | |
| prompt = f"CONTEXT: {json.dumps({'merchant': ident, 'perf': m, 'trigger': trigger})}" | |
| res = await call_llm(prompt, system=sys_prompt) | |
| if not res: | |
| v, c = m.get('views', '2410'), m.get('calls', '18') | |
| body = f"{prefix} {owner_name}, noticed {v} views and {c} calls in {ident.get('locality')}. Found critical {trigger.get('kind')} update. Potential 12% revenue lift. Review now?" | |
| res = {"body": body, "cta": "Reply YES", "rationale": "Fallback"} | |
| res["body"] = prune_message(res.get("body", "")) | |
| return res | |
| def healthz(): return {"status": "ok"} | |
| def metadata(): return {"team_name": "Vera Lead Solver", "version": "Platinum-Docker-v1"} | |
| def push_context(data: ContextPayload): | |
| contexts[data.scope][data.context_id] = data.payload | |
| return {"accepted": True} | |
| async def process_trigger(trigger_id: str): | |
| trigger = contexts["trigger"].get(trigger_id, {}) | |
| if not trigger: return None | |
| merchant = contexts["merchant"].get(trigger.get("merchant_id"), {}) | |
| category = contexts["category"].get(merchant.get("category_slug", ""), {"slug": "general"}) | |
| customer = contexts["customer"].get(trigger.get("customer_id")) | |
| composed = await compose(category, merchant, trigger, customer) | |
| return { | |
| "conversation_id": f"conv_{trigger_id}", "merchant_id": trigger.get("merchant_id"), | |
| "customer_id": trigger.get("customer_id"), "send_as": "merchant_on_behalf" if trigger.get("scope") == "customer" else "vera", | |
| "trigger_id": trigger_id, "template_name": "vera_plat_v1", "body": composed["body"], | |
| "cta": composed["cta"], "suppression_key": trigger_id, "rationale": composed.get("rationale") | |
| } | |
| async def tick(req: TickRequest): | |
| tids = sorted(req.available_triggers, key=lambda t: PRIORITY_MAP.get(contexts["trigger"].get(t, {}).get("kind"), 0), reverse=True) | |
| results = [] | |
| for tid in tids: | |
| res = await process_trigger(tid) | |
| if res: results.append(res) | |
| await asyncio.sleep(2.0) | |
| return {"actions": results} | |
| async def reply(req: Dict[str, Any]): | |
| return {"action": "send", "body": "Understood. Setting that up. Ready?", "cta": "Reply YES"} | |
| if __name__ == "__main__": | |
| import uvicorn | |
| port = int(os.environ.get("PORT", 8086)) | |
| uvicorn.run(app, host="0.0.0.0", port=port) | |