diff --git "a/bios_controller.py" "b/bios_controller.py" --- "a/bios_controller.py" +++ "b/bios_controller.py" @@ -1,1190 +1,1273 @@ -""" -╔══════════════════════════════════════════════════════════════════════════════╗ -║ ║ -║ BIOS — Business Idea Operating System ║ -║ Model Controller · bios_controller.py ║ -║ Version: 1.0.0 · Kernel: BIOS-kernel-v1 ║ -║ ║ -║ "We don't just analyse businesses. We illuminate them." ║ -║ ║ -╚══════════════════════════════════════════════════════════════════════════════╝ - -Architecture: - BIOSController - ├── ModelRouter — switches between base LLM and BIOS-Insight-v1 - ├── DiagnosisEngine — processes 24 questions, runs health score formula - ├── InsightGenerator — builds structured JSON diagnosis report - └── NeonDBWriter — persists results to PostgreSQL via psycopg -""" - -from __future__ import annotations - -import json -import logging -import os -import re -import time -import uuid -from dataclasses import dataclass, field, asdict -from datetime import datetime, timezone -from enum import Enum -from typing import Any, Optional - -import psycopg # psycopg v3 (pip install psycopg[binary]) -from psycopg.rows import dict_row - -# ── Optional: HuggingFace Inference (pip install huggingface_hub) ────────── -try: - from huggingface_hub import InferenceClient - HF_AVAILABLE = True -except ImportError: - HF_AVAILABLE = False - -# ── Optional: Groq client for llama-3.3-70b (pip install groq) ───────────── -try: - from groq import Groq - GROQ_AVAILABLE = True -except ImportError: - GROQ_AVAILABLE = False - -# ── Optional: Anthropic (pip install anthropic) ──────────────────────────── -try: - import anthropic - ANTHROPIC_AVAILABLE = True -except ImportError: - ANTHROPIC_AVAILABLE = False - - -# ═══════════════════════════════════════════════════════════════════════════════ -# LOGGING -# ═══════════════════════════════════════════════════════════════════════════════ - -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s [BIOS-%(levelname)s] %(message)s", - datefmt="%Y-%m-%d %H:%M:%S", -) -log = logging.getLogger("bios.controller") - - -# ═══════════════════════════════════════════════════════════════════════════════ -# ENUMS & CONSTANTS -# ═══════════════════════════════════════════════════════════════════════════════ - -class ModelBackend(str, Enum): - """Supported inference backends.""" - GROQ = "groq" # llama-3.3-70b-versatile via Groq - HF_INFERENCE = "hf_inference" # HuggingFace Inference API - ANTHROPIC = "anthropic" # Claude fallback - LOCAL = "local" # Local transformers pipeline - MOCK = "mock" # Offline / testing - - -class ModelVariant(str, Enum): - """Which model to route to.""" - BASE = "base" # General LLM (llama-3.3-70b) - BIOS_INSIGHT = "bios_insight" # Fine-tuned BIOS-Insight-v1 - - -# Model identifiers -MODEL_IDS = { - ModelVariant.BASE: "meta-llama/llama-3.3-70b-versatile", - ModelVariant.BIOS_INSIGHT: "BIOS-kernel/BIOS-Insight-v1", # future HF repo -} - -GROQ_MODEL_IDS = { - ModelVariant.BASE: "llama-3.3-70b-versatile", - ModelVariant.BIOS_INSIGHT: "llama-3.3-70b-versatile", # until HF model is live -} - -# Industry benchmarks (Myanmar SME context, values in MMK) -INDUSTRY_BENCHMARKS: dict[str, dict] = { - "Gold Shop": {"avg_revenue": 15_000_000, "avg_retention": 60, "avg_clv": 2_000_000, "avg_team": 4, "avg_mkt": 200_000}, - "Fashion": {"avg_revenue": 8_000_000, "avg_retention": 40, "avg_clv": 300_000, "avg_team": 6, "avg_mkt": 500_000}, - "F&B": {"avg_revenue": 10_000_000, "avg_retention": 50, "avg_clv": 150_000, "avg_team": 10, "avg_mkt": 400_000}, - "Cosmetics": {"avg_revenue": 6_000_000, "avg_retention": 45, "avg_clv": 250_000, "avg_team": 5, "avg_mkt": 600_000}, - "Electronics": {"avg_revenue": 20_000_000, "avg_retention": 35, "avg_clv": 800_000, "avg_team": 8, "avg_mkt": 700_000}, - "Other": {"avg_revenue": 5_000_000, "avg_retention": 40, "avg_clv": 200_000, "avg_team": 5, "avg_mkt": 300_000}, -} - - -# ═══════════════════════════════════════════════════════════════════════════════ -# DATA MODELS -# ═══════════════════════════════════════════════════════════════════════════════ - -@dataclass -class BusinessInputs: - """ - Complete set of 24 diagnostic questions, grouped into 4 sections. - - All monetary values are in MMK (Myanmar Kyat). - Percentages are 0–100 (e.g. retention_rate=65 means 65%). - """ - - # ── Section 1: Business Basics (6 questions) ────────────────────────────── - business_name: str = "" # Q1 - industry: str = "Other" # Q2 Gold Shop / Fashion / F&B / Cosmetics / Electronics / Other - location: str = "Yangon" # Q3 Yangon / Mandalay / Naypyidaw / Other - years_in_business: int = 0 # Q4 0–100 - monthly_revenue: float = 0.0 # Q5 MMK - team_size: int = 1 # Q6 headcount - - # ── Section 2: Market & Customers (6 questions) ─────────────────────────── - target_customer: str = "" # Q7 free text - acquisition_channels: list[str] = field(default_factory=list) # Q8 multi-select - avg_customer_lifetime_value: float = 0.0 # Q9 MMK - retention_rate: float = 0.0 # Q10 % - main_competitors: str = "" # Q11 optional - unique_selling_proposition: str = "" # Q12 - - # ── Section 3: Operations & Challenges (6 questions) ───────────────────── - sales_channels: list[str] = field(default_factory=list) # Q13 - operational_challenge: str = "" # Q14 - biggest_pain_point: str = "" # Q15 - current_technology: list[str] = field(default_factory=list) # Q16 - marketing_channels: list[str] = field(default_factory=list) # Q17 - monthly_marketing_budget: float = 0.0 # Q18 MMK - - # ── Section 4: Goals & Constraints (6 questions) ───────────────────────── - goal_3_month: float = 0.0 # Q19 MMK - goal_6_month: float = 0.0 # Q20 MMK - goal_12_month: float = 0.0 # Q21 MMK - budget_constraint: str = "Moderate (200-500K)" # Q22 - tech_readiness: str = "Somewhat ready" # Q23 - preferred_language: str = "English" # Q24 - - -@dataclass -class HealthDimensions: - """Sub-scores for the five health dimensions (each 0–100).""" - revenue_strength: int = 0 - customer_retention: int = 0 - market_position: int = 0 - technology_adoption: int = 0 - growth_trajectory: int = 0 - - @property - def total(self) -> int: - """ - Official BIOS Health Score formula: - (Revenue Strength × 20) + (Customer Retention × 20) + - (Market Position × 20) + (Technology Adoption × 20) + - (Growth Trajectory × 20) - Each dimension is 0–100, weight is 20%, so max = 100. - """ - return round( - (self.revenue_strength * 0.20) + - (self.customer_retention * 0.20) + - (self.market_position * 0.20) + - (self.technology_adoption * 0.20) + - (self.growth_trajectory * 0.20) - ) - - def to_dict(self) -> dict: - return { - "revenue_strength": self.revenue_strength, - "customer_retention": self.customer_retention, - "market_position": self.market_position, - "technology_adoption": self.technology_adoption, - "growth_trajectory": self.growth_trajectory, - "total": self.total, - } - - -@dataclass -class Weakness: - rank: int - dimension: str - label: str - your_score: float - benchmark: float - gap: float - severity: str # HIGH / MEDIUM / LOW - detail: str - - def to_dict(self) -> dict: - return asdict(self) - - -@dataclass -class Opportunity: - rank: int - title: str - description: str - expected_impact: str - difficulty: str # EASY / MEDIUM / HARD - timeframe: str - revenue_uplift_mmk: Optional[float] = None - - def to_dict(self) -> dict: - return asdict(self) - - -@dataclass -class ActionItem: - priority: int - action: str - rationale: str - urgency_score: float - impact_score: float - feasibility_score: float - composite_score: float - - def to_dict(self) -> dict: - return asdict(self) - - -@dataclass -class DiagnosisReport: - """Full Module 1 output — the BIOS diagnosis report.""" - session_id: str - business_name: str - industry: str - location: str - generated_at: str - - health_score: int - health_label: str # Critical / Below Average / Fair / Good / Excellent - health_dimensions: HealthDimensions - - top_3_weaknesses: list[Weakness] - growth_opportunities: list[Opportunity] - priority_action_items: list[ActionItem] - - ai_narrative: str # BIOS LLM executive summary - benchmarking: list[dict] - next_module: str = "Strategy Engine (Module 2)" - - model_used: str = "" - generation_time_ms: int = 0 - - def to_dict(self) -> dict: - return { - "session_id": self.session_id, - "business_name": self.business_name, - "industry": self.industry, - "location": self.location, - "generated_at": self.generated_at, - "health_score": self.health_score, - "health_label": self.health_label, - "health_dimensions": self.health_dimensions.to_dict(), - "top_3_weaknesses": [w.to_dict() for w in self.top_3_weaknesses], - "growth_opportunities": [o.to_dict() for o in self.growth_opportunities], - "priority_action_items": [a.to_dict() for a in self.priority_action_items], - "ai_narrative": self.ai_narrative, - "benchmarking": self.benchmarking, - "next_module": self.next_module, - "model_used": self.model_used, - "generation_time_ms": self.generation_time_ms, - } - - def to_json(self, indent: int = 2) -> str: - return json.dumps(self.to_dict(), ensure_ascii=False, indent=indent) - - -# ═══════════════════════════════════════════════════════════════════════════════ -# MODEL ROUTER -# ═══════════════════════════════════════════════════════════════════════════════ - -class ModelRouter: - """ - Routes inference requests to the appropriate backend + model variant. - - Priority order when calling .infer(): - 1. If BIOS-Insight-v1 is flagged as available → use HF Inference API - 2. Else use base model via Groq (fastest, free tier) - 3. Fallback to Anthropic Claude - 4. Final fallback: MOCK mode (returns structured placeholder) - """ - - def __init__( - self, - backend: ModelBackend = ModelBackend.GROQ, - variant: ModelVariant = ModelVariant.BASE, - bios_insight_ready: bool = False, - temperature: float = 0.3, - max_tokens: int = 2048, - ): - self.backend = backend - self.variant = variant - self.bios_insight_ready = bios_insight_ready - self.temperature = temperature - self.max_tokens = max_tokens - - # Clients initialised lazily - self._groq_client: Any = None - self._hf_client: Any = None - self._anthropic_client: Any = None - - log.info( - f"ModelRouter initialised | backend={backend.value} " - f"variant={variant.value} | BIOS-Insight-v1 ready={bios_insight_ready}" - ) - - # ── Client factories ────────────────────────────────────────────────────── - - def _get_groq(self): - if self._groq_client is None: - if not GROQ_AVAILABLE: - raise RuntimeError("groq package not installed. Run: pip install groq") - api_key = os.getenv("GROQ_API_KEY") - if not api_key: - raise RuntimeError("GROQ_API_KEY environment variable not set") - self._groq_client = Groq(api_key=api_key) - return self._groq_client - - def _get_hf(self): - if self._hf_client is None: - if not HF_AVAILABLE: - raise RuntimeError("huggingface_hub not installed. Run: pip install huggingface_hub") - api_key = os.getenv("HF_API_KEY") - if not api_key: - raise RuntimeError("HF_API_KEY environment variable not set") - self._hf_client = InferenceClient(token=api_key) - return self._hf_client - - def _get_anthropic(self): - if self._anthropic_client is None: - if not ANTHROPIC_AVAILABLE: - raise RuntimeError("anthropic package not installed. Run: pip install anthropic") - api_key = os.getenv("ANTHROPIC_API_KEY") - if not api_key: - raise RuntimeError("ANTHROPIC_API_KEY not set") - self._anthropic_client = anthropic.Anthropic(api_key=api_key) - return self._anthropic_client - - # ── Routing decision ────────────────────────────────────────────────────── - - def _resolve_route(self) -> tuple[ModelBackend, ModelVariant]: - """Determine which backend + variant to actually use.""" - if self.bios_insight_ready and HF_AVAILABLE and os.getenv("HF_API_KEY"): - return ModelBackend.HF_INFERENCE, ModelVariant.BIOS_INSIGHT - if GROQ_AVAILABLE and os.getenv("GROQ_API_KEY"): - return ModelBackend.GROQ, ModelVariant.BASE - if ANTHROPIC_AVAILABLE and os.getenv("ANTHROPIC_API_KEY"): - return ModelBackend.ANTHROPIC, ModelVariant.BASE - return ModelBackend.MOCK, ModelVariant.BASE - - # ── Inference ───────────────────────────────────────────────────────────── - - def infer(self, system_prompt: str, user_prompt: str) -> tuple[str, str]: - """ - Send prompts to the resolved model. - - Returns: - (response_text, model_identifier_used) - """ - backend, variant = _resolve_route(self) if False else self._resolve_route() - model_id = GROQ_MODEL_IDS.get(variant, GROQ_MODEL_IDS[ModelVariant.BASE]) - log.info(f"Routing → backend={backend.value} model={model_id}") - - if backend == ModelBackend.GROQ: - return self._infer_groq(system_prompt, user_prompt, model_id) - - if backend == ModelBackend.HF_INFERENCE: - hf_model = MODEL_IDS[ModelVariant.BIOS_INSIGHT] - return self._infer_hf(system_prompt, user_prompt, hf_model) - - if backend == ModelBackend.ANTHROPIC: - return self._infer_anthropic(system_prompt, user_prompt) - - # MOCK fallback - return self._mock_response(), "mock/bios-kernel-v1" - - def _infer_groq(self, system: str, user: str, model: str) -> tuple[str, str]: - client = self._get_groq() - response = client.chat.completions.create( - model=model, - messages=[ - {"role": "system", "content": system}, - {"role": "user", "content": user}, - ], - temperature=self.temperature, - max_tokens=self.max_tokens, - response_format={"type": "json_object"}, - ) - return response.choices[0].message.content, f"groq/{model}" - - def _infer_hf(self, system: str, user: str, model: str) -> tuple[str, str]: - client = self._get_hf() - messages = [ - {"role": "system", "content": system}, - {"role": "user", "content": user}, - ] - response = client.chat_completion( - messages=messages, - model=model, - max_tokens=self.max_tokens, - temperature=self.temperature, - ) - return response.choices[0].message.content, f"hf/{model}" - - def _infer_anthropic(self, system: str, user: str) -> tuple[str, str]: - client = self._get_anthropic() - message = client.messages.create( - model="claude-sonnet-4-20250514", - max_tokens=self.max_tokens, - system=system, - messages=[{"role": "user", "content": user}], - ) - return message.content[0].text, "anthropic/claude-sonnet-4-20250514" - - def _mock_response(self) -> str: - """Return a valid JSON mock for offline testing.""" - return json.dumps({ - "narrative": ( - "BIOS analysis complete. Your business shows strong foundational " - "elements but faces challenges in customer retention and technology " - "adoption. Prioritise loyalty initiatives and digital tooling to " - "unlock the next growth tier." - ), - "model": "mock", - }) - - -# ═══════════════════════════════════════════════════════════════════════════════ -# DIAGNOSIS ENGINE (pure scoring logic — no LLM required) -# ═══════════════════════════════════════════════════════════════════════════════ - -class DiagnosisEngine: - """ - Implements the BIOS Module 1 scoring algorithms. - - All calculations are deterministic and reproducible — the LLM is only - used to generate the qualitative narrative on top of these numbers. - """ - - # ── Dimension scorers ───────────────────────────────────────────────────── - - @staticmethod - def score_revenue(monthly_revenue: float) -> int: - thresholds = [ - (50_000_000, 100), - (20_000_000, 80), - ( 5_000_000, 60), - ( 1_000_000, 40), - ] - for threshold, score in thresholds: - if monthly_revenue >= threshold: - return score - return 20 - - @staticmethod - def score_retention(rate: float) -> int: - thresholds = [(80, 100), (60, 80), (40, 60), (20, 40)] - for threshold, score in thresholds: - if rate >= threshold: - return score - return 20 - - @staticmethod - def score_market_position(usp: str, competitors: str) -> int: - words = len(usp.strip().split()) - base = 20 - if words >= 50: base = 80 - elif words >= 30: base = 60 - elif words >= 15: base = 40 - # Bonus for knowing your competition (+5, capped at 100) - if competitors and len(competitors.strip()) > 5: - base = min(100, base + 5) - return base - - @staticmethod - def score_technology(technology: list[str]) -> int: - tech_lower = [t.lower() for t in technology] - if not tech_lower or "none" in tech_lower: - return 10 - advanced = {"erp", "crm", "ai tools", "automation", "bi dashboard"} - mid_tier = {"pos system", "accounting software", "inventory system"} - basic = {"spreadsheets", "facebook business suite", "whatsapp business"} - if any(t in advanced for t in tech_lower): return 100 - if any(t in mid_tier for t in tech_lower): return 60 - if any(t in basic for t in tech_lower): return 30 - return 20 - - @staticmethod - def score_growth(goal_12: float, current: float) -> int: - if current <= 0: - return 40 # can't compute — neutral score - rate = (goal_12 - current) / current * 100 - thresholds = [(50, 100), (30, 80), (10, 60), (0, 40)] - for threshold, score in thresholds: - if rate >= threshold: - return score - return 20 - - # ── Main scorer ─────────────────────────────────────────────────────────── - - def compute_dimensions(self, inp: BusinessInputs) -> HealthDimensions: - return HealthDimensions( - revenue_strength = self.score_revenue(inp.monthly_revenue), - customer_retention = self.score_retention(inp.retention_rate), - market_position = self.score_market_position( - inp.unique_selling_proposition, - inp.main_competitors), - technology_adoption = self.score_technology(inp.current_technology), - growth_trajectory = self.score_growth( - inp.goal_12_month, - inp.monthly_revenue), - ) - - @staticmethod - def health_label(score: int) -> str: - if score >= 80: return "Excellent" - if score >= 65: return "Good" - if score >= 45: return "Fair" - if score >= 30: return "Below Average" - return "Critical" - - # ── Weakness identification ─────────────────────────────────────────────── - - def identify_weaknesses( - self, - inp: BusinessInputs, - dims: HealthDimensions, - ) -> list[Weakness]: - bench = INDUSTRY_BENCHMARKS.get(inp.industry, INDUSTRY_BENCHMARKS["Other"]) - bench_scores = { - "revenue_strength": self.score_revenue(bench["avg_revenue"]), - "customer_retention": self.score_retention(bench["avg_retention"]), - "market_position": 60, # industry-standard expectation - "technology_adoption": 60, - "growth_trajectory": 60, - } - labels = { - "revenue_strength": "Monthly Revenue", - "customer_retention": "Customer Retention", - "market_position": "Market Differentiation", - "technology_adoption": "Technology Adoption", - "growth_trajectory": "Growth Ambition", - } - details = { - "revenue_strength": f"Revenue of {inp.monthly_revenue:,.0f} MMK is significantly below the {inp.industry} industry average.", - "customer_retention": f"Only {inp.retention_rate:.0f}% repeat purchase rate — industry average is {bench['avg_retention']:.0f}%.", - "market_position": "Your unique selling proposition needs greater clarity and depth to stand out.", - "technology_adoption": "Low technology adoption is limiting operational efficiency and scalability.", - "growth_trajectory": "Growth goals are misaligned with current revenue trajectory.", - } - - user_scores = { - "revenue_strength": dims.revenue_strength, - "customer_retention": dims.customer_retention, - "market_position": dims.market_position, - "technology_adoption": dims.technology_adoption, - "growth_trajectory": dims.growth_trajectory, - } - - weaknesses: list[Weakness] = [] - for key, user_score in user_scores.items(): - b_score = bench_scores[key] - if user_score < b_score * 0.8: - gap = b_score - user_score - severity = "HIGH" if gap > 30 else "MEDIUM" if gap > 15 else "LOW" - weaknesses.append(Weakness( - rank=0, - dimension=key, - label=labels[key], - your_score=user_score, - benchmark=b_score, - gap=round(gap, 1), - severity=severity, - detail=details[key], - )) - - # Sort: HIGH first, then by gap descending - sev_order = {"HIGH": 3, "MEDIUM": 2, "LOW": 1} - weaknesses.sort(key=lambda w: (sev_order[w.severity], w.gap), reverse=True) - top3 = weaknesses[:3] - for i, w in enumerate(top3, 1): - w.rank = i - return top3 - - # ── Opportunity discovery ───────────────────────────────────────────────── - - def discover_opportunities( - self, - inp: BusinessInputs, - dims: HealthDimensions, - ) -> list[Opportunity]: - bench = INDUSTRY_BENCHMARKS.get(inp.industry, INDUSTRY_BENCHMARKS["Other"]) - opps: list[Opportunity] = [] - - # 1. Revenue growth - if inp.goal_12_month > inp.monthly_revenue and inp.monthly_revenue > 0: - pct = (inp.goal_12_month / inp.monthly_revenue - 1) * 100 - opps.append(Opportunity( - rank=0, - title="Scale Revenue Toward 12-Month Goal", - description=f"Bridge the {pct:.0f}% gap to your {inp.goal_12_month:,.0f} MMK annual revenue target.", - expected_impact=f"+{pct:.0f}% revenue growth", - difficulty="MEDIUM", - timeframe="6–12 months", - revenue_uplift_mmk=inp.goal_12_month - inp.monthly_revenue, - )) - - # 2. Retention improvement - if inp.retention_rate < bench["avg_retention"]: - gap = bench["avg_retention"] - inp.retention_rate - monthly_g = (gap / 100) * inp.avg_customer_lifetime_value * max(inp.team_size, 1) - opps.append(Opportunity( - rank=0, - title="Boost Customer Retention Rate", - description=( - f"Raise repeat-purchase rate by {gap:.0f}% to match the " - f"{inp.industry} industry benchmark." - ), - expected_impact=f"+{monthly_g:,.0f} MMK estimated monthly revenue", - difficulty="MEDIUM", - timeframe="2–3 months", - revenue_uplift_mmk=monthly_g * 12, - )) - - # 3. Channel expansion - if len(inp.sales_channels) < 3: - needed = 3 - len(inp.sales_channels) - opps.append(Opportunity( - rank=0, - title=f"Expand to {needed} New Sales Channel{'s' if needed > 1 else ''}", - description="Diversifying beyond your current channels reduces single-point risk and opens new customer pools.", - expected_impact="+20–30% customer reach", - difficulty="EASY", - timeframe="1–2 months", - )) - - # 4. Technology upgrade - if dims.technology_adoption < 60: - opps.append(Opportunity( - rank=0, - title="Adopt Core Business Technology", - description="Implementing a CRM or POS system unlocks data-driven decisions and staff efficiency.", - expected_impact="+15–25% operational efficiency", - difficulty="MEDIUM", - timeframe="2–4 weeks", - )) - - # 5. Marketing investment - if inp.monthly_marketing_budget < bench["avg_mkt"] * 0.5: - opps.append(Opportunity( - rank=0, - title="Increase Marketing Investment", - description=( - f"Current budget of {inp.monthly_marketing_budget:,.0f} MMK " - f"is far below the {inp.industry} average of {bench['avg_mkt']:,.0f} MMK." - ), - expected_impact="+10–20% new customer acquisition", - difficulty="EASY", - timeframe="1 month", - )) - - # Rank and cap at 5 - for i, opp in enumerate(opps[:5], 1): - opp.rank = i - return opps[:5] - - # ── Priority action items ───────────────────────────────────────────────── - - RECOMMENDED_ACTIONS: dict[str, str] = { - "revenue_strength": "Run a margin audit and introduce 2 high-value upsell products this month.", - "customer_retention": "Launch a loyalty stamp card and a 30-day follow-up WhatsApp message sequence.", - "market_position": "Rewrite your USP in one clear sentence and test it in Facebook ad copy.", - "technology_adoption": "Set up a free CRM (HubSpot or Zoho) and import your customer contact list.", - "growth_trajectory": "Break your 12-month target into monthly milestones and review weekly.", - } - - def rank_priority_actions( - self, - weaknesses: list[Weakness], - inp: BusinessInputs, - ) -> list[ActionItem]: - items: list[ActionItem] = [] - sev_urgency = {"HIGH": 85, "MEDIUM": 60, "LOW": 35} - - for w in weaknesses: - urgency = float(sev_urgency.get(w.severity, 40)) - impact = min(100.0, w.gap * 1.6) - feasibility = {"HIGH": 40.0, "MEDIUM": 65.0, "LOW": 80.0}.get(w.severity, 50.0) - composite = round(urgency * 0.4 + impact * 0.4 + feasibility * 0.2, 1) - - items.append(ActionItem( - priority=0, - action=self.RECOMMENDED_ACTIONS.get(w.dimension, f"Address {w.label} urgently."), - rationale=w.detail, - urgency_score=urgency, - impact_score=round(impact, 1), - feasibility_score=feasibility, - composite_score=composite, - )) - - items.sort(key=lambda x: x.composite_score, reverse=True) - for i, item in enumerate(items, 1): - item.priority = i - return items - - # ── Benchmarking table ──────────────────────────────────────────────────── - - def build_benchmarking( - self, - inp: BusinessInputs, - ) -> list[dict]: - bench = INDUSTRY_BENCHMARKS.get(inp.industry, INDUSTRY_BENCHMARKS["Other"]) - - def status(val: float, avg: float) -> str: - if val >= avg * 1.10: return "ABOVE" - if val <= avg * 0.90: return "BELOW" - return "AT" - - return [ - {"metric": "Monthly Revenue", "your_value": inp.monthly_revenue, "industry_avg": bench["avg_revenue"], "unit": "MMK", "status": status(inp.monthly_revenue, bench["avg_revenue"])}, - {"metric": "Customer Retention Rate", "your_value": inp.retention_rate, "industry_avg": bench["avg_retention"], "unit": "%", "status": status(inp.retention_rate, bench["avg_retention"])}, - {"metric": "Avg Customer Lifetime Val","your_value": inp.avg_customer_lifetime_value, "industry_avg": bench["avg_clv"], "unit": "MMK", "status": status(inp.avg_customer_lifetime_value, bench["avg_clv"])}, - {"metric": "Marketing Budget", "your_value": inp.monthly_marketing_budget, "industry_avg": bench["avg_mkt"], "unit": "MMK", "status": status(inp.monthly_marketing_budget, bench["avg_mkt"])}, - {"metric": "Team Size", "your_value": float(inp.team_size), "industry_avg": float(bench["avg_team"]),"unit": "ppl", "status": status(inp.team_size, bench["avg_team"])}, - ] - - -# ═══════════════════════════════════════════════════════════════════════════════ -# INSIGHT GENERATOR (builds the LLM prompt and parses the response) -# ═══════════════════════════════════════════════════════════════════════════════ - -class InsightGenerator: - """ - Constructs structured prompts for the BIOS LLM and parses its JSON output - into the qualitative `ai_narrative` field of the DiagnosisReport. - """ - - SYSTEM_PROMPT = """You are BIOS — the Business Idea Operating System. You are the elite AI advisor for Myanmar SMEs, Gold Shops, and ambitious entrepreneurs across Southeast Asia. - -Your personality: professional, precise, encouraging, and bold — like a McKinsey partner who speaks to founders, not just analysts. You use the Dark & Gold luxury tone: every word carries weight, every recommendation is actionable. - -You always respond in valid JSON with this exact structure: -{ - "narrative": "<3-paragraph executive summary in the user's preferred language>", - "headline_insight": "" -} - -Rules: -- Be specific with numbers from the data provided -- Use the language specified in preferred_language -- Never be generic — reference the actual business, industry, and goals -- Tone: elite advisory, not chatbot small talk""" - - def build_user_prompt(self, inp: BusinessInputs, dims: HealthDimensions, weaknesses: list[Weakness], opps: list[Opportunity]) -> str: - weak_lines = "\n".join( - f" {w.rank}. {w.label} — score {w.your_score:.0f} vs benchmark {w.benchmark:.0f} | severity: {w.severity}" - for w in weaknesses - ) - opp_lines = "\n".join( - f" {o.rank}. {o.title}: {o.expected_impact} ({o.timeframe})" - for o in opps[:3] - ) - - return f"""BIOS DIAGNOSIS DATA — analyse and generate insights. - -Business: {inp.business_name} -Industry: {inp.industry} | Location: {inp.location} -Years Operating: {inp.years_in_business} | Team: {inp.team_size} people -Monthly Revenue: {inp.monthly_revenue:,.0f} MMK -Retention Rate: {inp.retention_rate:.0f}% -Monthly Marketing Budget: {inp.monthly_marketing_budget:,.0f} MMK -Preferred Language: {inp.preferred_language} - -HEALTH SCORE: {dims.total}/100 — {DiagnosisEngine.health_label(dims.total)} - Revenue Strength: {dims.revenue_strength} - Customer Retention: {dims.customer_retention} - Market Position: {dims.market_position} - Technology Adoption: {dims.technology_adoption} - Growth Trajectory: {dims.growth_trajectory} - -TOP WEAKNESSES: -{weak_lines} - -TOP OPPORTUNITIES: -{opp_lines} - -12-Month Revenue Goal: {inp.goal_12_month:,.0f} MMK -Unique Value Proposition: "{inp.unique_selling_proposition}" -Biggest Pain Point: "{inp.biggest_pain_point}" - -Generate the executive narrative JSON now.""" - - def parse_narrative(self, raw: str) -> str: - """Extract narrative from LLM JSON response, with fallback.""" - try: - # Strip markdown fences if present - clean = re.sub(r"```(?:json)?\s*", "", raw).strip().rstrip("```").strip() - data = json.loads(clean) - return data.get("narrative", raw) - except (json.JSONDecodeError, AttributeError): - # Return raw text if not parseable - return raw.strip() - - -# ═══════════════════════════════════════════════════════════════════════════════ -# NEON DB WRITER -# ═══════════════════════════════════════════════════════════════════════════════ - -class NeonDBWriter: - """ - Persists BIOS diagnosis reports to NeonDB (PostgreSQL) via psycopg v3. - - Required table (run schema_auth.sql first): - CREATE TABLE IF NOT EXISTS diagnoses ( - id UUID PRIMARY KEY DEFAULT uuid_generate_v4(), - session_id VARCHAR(255) UNIQUE NOT NULL, - business_name VARCHAR(255), - industry VARCHAR(100), - location VARCHAR(100), - health_score INTEGER, - health_label VARCHAR(50), - health_dimensions JSONB, - top_3_weaknesses JSONB, - growth_opportunities JSONB, - priority_action_items JSONB, - ai_narrative TEXT, - benchmarking JSONB, - model_used VARCHAR(255), - generation_time_ms INTEGER, - status VARCHAR(50) DEFAULT 'COMPLETED', - created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() - ); - """ - - def __init__(self, database_url: Optional[str] = None): - self.database_url = database_url or os.getenv("DATABASE_URL") - if not self.database_url: - raise ValueError( - "DATABASE_URL not set. Export it or pass database_url= to NeonDBWriter." - ) - # Normalise asyncpg URL to psycopg URL - self.database_url = self.database_url.replace( - "postgresql+asyncpg://", "postgresql://" - ).replace( - "postgres+asyncpg://", "postgresql://" - ) - - def save_report(self, report: DiagnosisReport) -> str: - """ - Upsert a DiagnosisReport into the diagnoses table. - Returns the session_id of the saved record. - """ - d = report.to_dict() - - sql = """ - INSERT INTO diagnoses ( - session_id, business_name, industry, location, - health_score, health_label, health_dimensions, - top_3_weaknesses, growth_opportunities, priority_action_items, - ai_narrative, benchmarking, model_used, generation_time_ms, - status, created_at - ) VALUES ( - %(session_id)s, %(business_name)s, %(industry)s, %(location)s, - %(health_score)s, %(health_label)s, %(health_dimensions)s, - %(top_3_weaknesses)s, %(growth_opportunities)s, %(priority_action_items)s, - %(ai_narrative)s, %(benchmarking)s, %(model_used)s, %(generation_time_ms)s, - 'COMPLETED', NOW() - ) - ON CONFLICT (session_id) DO UPDATE SET - health_score = EXCLUDED.health_score, - health_label = EXCLUDED.health_label, - health_dimensions = EXCLUDED.health_dimensions, - top_3_weaknesses = EXCLUDED.top_3_weaknesses, - growth_opportunities = EXCLUDED.growth_opportunities, - priority_action_items = EXCLUDED.priority_action_items, - ai_narrative = EXCLUDED.ai_narrative, - benchmarking = EXCLUDED.benchmarking, - model_used = EXCLUDED.model_used, - generation_time_ms = EXCLUDED.generation_time_ms, - status = 'COMPLETED' - """ - - params = { - "session_id": d["session_id"], - "business_name": d["business_name"], - "industry": d["industry"], - "location": d["location"], - "health_score": d["health_score"], - "health_label": d["health_label"], - "health_dimensions": json.dumps(d["health_dimensions"]), - "top_3_weaknesses": json.dumps(d["top_3_weaknesses"]), - "growth_opportunities": json.dumps(d["growth_opportunities"]), - "priority_action_items": json.dumps(d["priority_action_items"]), - "ai_narrative": d["ai_narrative"], - "benchmarking": json.dumps(d["benchmarking"]), - "model_used": d["model_used"], - "generation_time_ms": d["generation_time_ms"], - } - - with psycopg.connect(self.database_url) as conn: - with conn.cursor() as cur: - cur.execute(sql, params) - conn.commit() - - log.info(f"✅ Report saved to NeonDB | session_id={report.session_id} | score={report.health_score}") - return report.session_id - - def fetch_report(self, session_id: str) -> Optional[dict]: - """Fetch a previously saved report by session_id.""" - sql = "SELECT * FROM diagnoses WHERE session_id = %s" - with psycopg.connect(self.database_url, row_factory=dict_row) as conn: - with conn.cursor() as cur: - cur.execute(sql, (session_id,)) - row = cur.fetchone() - return dict(row) if row else None - - def list_reports(self, limit: int = 20) -> list[dict]: - """Return the most recent diagnoses.""" - sql = "SELECT session_id, business_name, industry, health_score, health_label, status, created_at FROM diagnoses ORDER BY created_at DESC LIMIT %s" - with psycopg.connect(self.database_url, row_factory=dict_row) as conn: - with conn.cursor() as cur: - cur.execute(sql, (limit,)) - rows = cur.fetchall() - return [dict(r) for r in rows] - - -# ═══════════════════════════════════════════════════════════════════════════════ -# BIOS CONTROLLER — the main entry point -# ═══════════════════════════════════════════════════════════════════════════════ - -class BIOSController: - """ - BIOS-kernel-v1 · Business Idea Operating System · Module 1 Controller - - Orchestrates the full diagnosis pipeline: - inputs → scoring → LLM narrative → structured report → NeonDB - - Usage: - controller = BIOSController() - report = controller.run_diagnosis(inputs) - print(report.to_json()) - - To use the fine-tuned BIOS-Insight-v1 when it becomes available on HF: - controller = BIOSController(bios_insight_ready=True) - """ - - VERSION = "1.0.0" - KERNEL = "BIOS-kernel-v1" - - def __init__( - self, - backend: ModelBackend = ModelBackend.GROQ, - bios_insight_ready: bool = False, - temperature: float = 0.3, - max_tokens: int = 2048, - database_url: Optional[str] = None, - save_to_db: bool = True, - ): - self.router = ModelRouter( - backend=backend, - bios_insight_ready=bios_insight_ready, - temperature=temperature, - max_tokens=max_tokens, - ) - self.engine = DiagnosisEngine() - self.generator = InsightGenerator() - self.save_to_db = save_to_db - - # DB writer — lazy init so missing DB URL doesn't break non-DB usage - self._db: Optional[NeonDBWriter] = None - self._db_url = database_url - - log.info( - f"╔═══════════════════════════════════════════╗\n" - f" BIOS Controller v{self.VERSION} · {self.KERNEL}\n" - f" backend={backend.value} | save_db={save_to_db}\n" - f"╚═══════════════════════════════════════════╝" - ) - - @property - def db(self) -> NeonDBWriter: - if self._db is None: - self._db = NeonDBWriter(self._db_url) - return self._db - - # ── Main pipeline ───────────────────────────────────────────────────────── - - def run_diagnosis(self, inputs: BusinessInputs) -> DiagnosisReport: - """ - Full Module 1 pipeline. - - Args: - inputs: Completed BusinessInputs with all 24 question answers. - - Returns: - DiagnosisReport with health_score, top_3_weaknesses, - growth_opportunities, priority_action_items, and ai_narrative. - """ - t_start = time.perf_counter() - session_id = str(uuid.uuid4()) - - log.info(f"▶ Starting BIOS diagnosis | business='{inputs.business_name}' | session={session_id}") - - # ── Step 1: Compute health dimensions and score ────────────────────── - dims = self.engine.compute_dimensions(inputs) - score = dims.total - label = self.engine.health_label(score) - log.info(f" Health Score: {score}/100 ({label})") - log.info(f" Dimensions: Rev={dims.revenue_strength} Ret={dims.customer_retention} Mkt={dims.market_position} Tech={dims.technology_adoption} Grow={dims.growth_trajectory}") - - # ── Step 2: Identify weaknesses ────────────────────────────────────── - weaknesses = self.engine.identify_weaknesses(inputs, dims) - log.info(f" Weaknesses identified: {[w.label for w in weaknesses]}") - - # ── Step 3: Discover opportunities ─────────────────────────────────── - opportunities = self.engine.discover_opportunities(inputs, dims) - log.info(f" Opportunities found: {len(opportunities)}") - - # ── Step 4: Rank priority actions ──────────────────────────────────── - actions = self.engine.rank_priority_actions(weaknesses, inputs) - - # ── Step 5: Build benchmarking table ───────────────────────────────── - benchmarking = self.engine.build_benchmarking(inputs) - - # ── Step 6: Generate AI narrative via LLM ──────────────────────────── - narrative, model_used = self._generate_narrative(inputs, dims, weaknesses, opportunities) - log.info(f" Narrative generated | model={model_used}") - - t_ms = int((time.perf_counter() - t_start) * 1000) - - # ── Step 7: Assemble report ─────────────────────────────────────────── - report = DiagnosisReport( - session_id = session_id, - business_name = inputs.business_name, - industry = inputs.industry, - location = inputs.location, - generated_at = datetime.now(timezone.utc).isoformat(), - health_score = score, - health_label = label, - health_dimensions = dims, - top_3_weaknesses = weaknesses, - growth_opportunities = opportunities, - priority_action_items= actions, - ai_narrative = narrative, - benchmarking = benchmarking, - model_used = model_used, - generation_time_ms = t_ms, - ) - - log.info(f"✔ Diagnosis complete | score={score} | {t_ms}ms") - - # ── Step 8: Save to NeonDB ──────────────────────────────────────────── - if self.save_to_db: - try: - self.db.save_report(report) - except Exception as e: - log.warning(f"DB save failed (non-fatal): {e}") - - return report - - def _generate_narrative( - self, - inp: BusinessInputs, - dims: HealthDimensions, - weak: list[Weakness], - opps: list[Opportunity], - ) -> tuple[str, str]: - """Call the LLM and return (narrative_text, model_identifier).""" - system = self.generator.SYSTEM_PROMPT - user = self.generator.build_user_prompt(inp, dims, weak, opps) - try: - raw, model_id = self.router.infer(system, user) - narrative = self.generator.parse_narrative(raw) - return narrative, model_id - except Exception as e: - log.warning(f"LLM call failed, using fallback narrative: {e}") - fallback = ( - f"{inp.business_name} received a BIOS Health Score of {dims.total}/100 ({self.engine.health_label(dims.total)}). " - f"Key areas for immediate attention: {', '.join(w.label for w in weak[:2])}. " - f"Top opportunity: {opps[0].title if opps else 'revenue diversification'}." - ) - return fallback, "fallback/static" - - # ── Convenience helpers ─────────────────────────────────────────────────── - - def switch_to_bios_insight(self): - """Activate BIOS-Insight-v1 once it is published on HuggingFace.""" - self.router.bios_insight_ready = True - self.router.variant = ModelVariant.BIOS_INSIGHT - log.info("🌟 Switched to BIOS-Insight-v1 (fine-tuned model)") - - def switch_to_base(self): - """Revert to base llama-3.3-70b model.""" - self.router.bios_insight_ready = False - self.router.variant = ModelVariant.BASE - log.info("Reverted to base model (llama-3.3-70b)") - - def get_report(self, session_id: str) -> Optional[dict]: - """Retrieve a saved report from NeonDB.""" - return self.db.fetch_report(session_id) - - def list_reports(self, limit: int = 20) -> list[dict]: - """List recent diagnosis reports from NeonDB.""" - return self.db.list_reports(limit) - - -# ═══════════════════════════════════════════════════════════════════════════════ -# CLI / DEMO RUNNER -# ═══════════════════════════════════════════════════════════════════════════════ - -def _demo_inputs() -> BusinessInputs: - """Sample Gold Shop business for demonstration.""" - return BusinessInputs( - # Section 1 - business_name = "Shwe Zin Gold & Jewellery", - industry = "Gold Shop", - location = "Yangon", - years_in_business = 7, - monthly_revenue = 4_200_000, - team_size = 3, - # Section 2 - target_customer = "Middle-income families aged 30–55 in Yangon, buying gold for investment and gifting during festivals.", - acquisition_channels = ["Word-of-mouth", "Facebook", "Walk-in"], - avg_customer_lifetime_value = 350_000, - retention_rate = 28.0, - main_competitors = "Dagon Gold, KBZ Gems", - unique_selling_proposition = "We sell certified 99.9% pure gold at transparent prices with a 10-year buyback guarantee.", - # Section 3 - sales_channels = ["Physical Store", "Facebook"], - operational_challenge = "Inventory management", - biggest_pain_point = "Customers don't come back after the first purchase — we have no system to follow up.", - current_technology = ["Spreadsheets"], - marketing_channels = ["Facebook", "Word-of-mouth"], - monthly_marketing_budget= 80_000, - # Section 4 - goal_3_month = 5_500_000, - goal_6_month = 7_000_000, - goal_12_month = 12_000_000, - budget_constraint = "Tight (50-200K)", - tech_readiness = "Somewhat ready", - preferred_language = "English", - ) - - -if __name__ == "__main__": - print("\n" + "═" * 60) - print(" BIOS — Business Idea Operating System") - print(" BIOS-kernel-v1 · Module 1: Business Diagnosis") - print("═" * 60 + "\n") - - # ── Instantiate controller ──────────────────────────────────────────────── - # Set save_to_db=True and export DATABASE_URL to persist to NeonDB. - controller = BIOSController( - backend = ModelBackend.GROQ, - save_to_db = bool(os.getenv("DATABASE_URL")), - ) - - # ── Run diagnosis ───────────────────────────────────────────────────────── - inputs = _demo_inputs() - report = controller.run_diagnosis(inputs) - - # ── Print structured JSON output ────────────────────────────────────────── - print("\n" + "─" * 60) - print(" BIOS DIAGNOSIS REPORT") - print("─" * 60) - print(report.to_json()) - - print("\n" + "═" * 60) - print(f" Health Score : {report.health_score}/100 ({report.health_label})") - print(f" Session ID : {report.session_id}") - print(f" Model Used : {report.model_used}") - print(f" Generated in : {report.generation_time_ms}ms") - print("═" * 60 + "\n") +""" +╔══════════════════════════════════════════════════════════════════════════════╗ +║ ║ +║ BIOS — Business Idea Operating System ║ +║ Model Controller · bios_controller.py ║ +║ Version: 1.0.0 · Kernel: BIOS-kernel-v1 ║ +║ ║ +║ "We don't just analyse businesses. We illuminate them." ║ +║ ║ +╚══════════════════════════════════════════════════════════════════════════════╝ + +Architecture: + BIOSController + ├── ModelRouter — switches between base LLM and BIOS-Insight-v1 + ├── DiagnosisEngine — processes 24 questions, runs health score formula + ├── InsightGenerator — builds structured JSON diagnosis report + └── NeonDBWriter — persists results to PostgreSQL via psycopg +""" + +from __future__ import annotations + +import json +import logging +import os +import re +import time +import uuid +from dataclasses import dataclass, field, asdict +from datetime import datetime, timezone +from enum import Enum +from typing import Any, Optional + +import psycopg # psycopg v3 (pip install psycopg[binary]) +from psycopg.rows import dict_row + +# ── Optional: HuggingFace Inference (pip install huggingface_hub) ────────── +try: + from huggingface_hub import InferenceClient + HF_AVAILABLE = True +except ImportError: + HF_AVAILABLE = False + +# ── Optional: Groq client for llama-3.3-70b (pip install groq) ───────────── +try: + from groq import Groq + GROQ_AVAILABLE = True +except ImportError: + GROQ_AVAILABLE = False + +# ── Optional: OpenAI (pip install openai) ──────────────────────────── +try: + import openai + OPENAI_AVAILABLE = True +except ImportError: + OPENAI_AVAILABLE = False + +# ── Optional: Gemini (pip install google-generativeai) ─────────────── +try: + import google.generativeai as genai + GEMINI_AVAILABLE = True +except ImportError: + GEMINI_AVAILABLE = False + +# ── Optional: Anthropic (pip install anthropic) ──────────────────────────── +try: + import anthropic + ANTHROPIC_AVAILABLE = True +except ImportError: + ANTHROPIC_AVAILABLE = False + + +# ═══════════════════════════════════════════════════════════════════════════════ +# LOGGING +# ═══════════════════════════════════════════════════════════════════════════════ + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [BIOS-%(levelname)s] %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", +) +log = logging.getLogger("bios.controller") + + +# ═══════════════════════════════════════════════════════════════════════════════ +# ENUMS & CONSTANTS +# ═══════════════════════════════════════════════════════════════════════════════ + +class ModelBackend(str, Enum): + """Supported inference backends.""" + GROQ = "groq" # llama-3.3-70b-versatile via Groq + HF_INFERENCE = "hf_inference" # HuggingFace Inference API + OPENAI = "openai" # GPT-4, GPT-3.5 via OpenAI + GEMINI = "gemini" # Gemini Pro via Google + ANTHROPIC = "anthropic" # Claude fallback + LOCAL = "local" # Local transformers pipeline + MOCK = "mock" # Offline / testing + + +class ModelVariant(str, Enum): + """Which model to route to.""" + BASE = "base" # General LLM (llama-3.3-70b) + BIOS_INSIGHT = "bios_insight" # Fine-tuned BIOS-Insight-v1 + + +# Model identifiers +MODEL_IDS = { + ModelVariant.BASE: "meta-llama/llama-3.3-70b-versatile", + ModelVariant.BIOS_INSIGHT: "BIOS-kernel/BIOS-Insight-v1", # future HF repo +} + +GROQ_MODEL_IDS = { + ModelVariant.BASE: "llama-3.3-70b-versatile", + ModelVariant.BIOS_INSIGHT: "llama-3.3-70b-versatile", # until HF model is live +} + +# Industry benchmarks (Myanmar SME context, values in MMK) +INDUSTRY_BENCHMARKS: dict[str, dict] = { + "Gold Shop": {"avg_revenue": 15_000_000, "avg_retention": 60, "avg_clv": 2_000_000, "avg_team": 4, "avg_mkt": 200_000}, + "Fashion": {"avg_revenue": 8_000_000, "avg_retention": 40, "avg_clv": 300_000, "avg_team": 6, "avg_mkt": 500_000}, + "F&B": {"avg_revenue": 10_000_000, "avg_retention": 50, "avg_clv": 150_000, "avg_team": 10, "avg_mkt": 400_000}, + "Cosmetics": {"avg_revenue": 6_000_000, "avg_retention": 45, "avg_clv": 250_000, "avg_team": 5, "avg_mkt": 600_000}, + "Electronics": {"avg_revenue": 20_000_000, "avg_retention": 35, "avg_clv": 800_000, "avg_team": 8, "avg_mkt": 700_000}, + "Technology Startup": {"avg_revenue": 25_000_000, "avg_retention": 70, "avg_clv": 3_000_000, "avg_team": 12, "avg_mkt": 1_000_000}, + "Other": {"avg_revenue": 5_000_000, "avg_retention": 40, "avg_clv": 200_000, "avg_team": 5, "avg_mkt": 300_000}, +} + + +# ═══════════════════════════════════════════════════════════════════════════════ +# DATA MODELS +# ═══════════════════════════════════════════════════════════════════════════════ + +@dataclass +class BusinessInputs: + """ + Complete set of 24 diagnostic questions, grouped into 4 sections. + + All monetary values are in MMK (Myanmar Kyat). + Percentages are 0–100 (e.g. retention_rate=65 means 65%). + """ + + # ── Section 1: Business Basics (6 questions) ────────────────────────────── + business_name: str = "" # Q1 + industry: str = "Other" # Q2 Gold Shop / Fashion / F&B / Cosmetics / Electronics / Technology Startup / Other + location: str = "Yangon" # Q3 Yangon / Mandalay / Naypyidaw / Other + years_in_business: int = 0 # Q4 0–100 + monthly_revenue: float = 0.0 # Q5 MMK + team_size: int = 1 # Q6 headcount + + # ── Section 2: Market & Customers (6 questions) ─────────────────────────── + target_customer: str = "" # Q7 free text + acquisition_channels: list[str] = field(default_factory=list) # Q8 multi-select + avg_customer_lifetime_value: float = 0.0 # Q9 MMK + retention_rate: float = 0.0 # Q10 % + main_competitors: str = "" # Q11 optional + unique_selling_proposition: str = "" # Q12 + + # ── Section 3: Operations & Challenges (6 questions) ───────────────────── + sales_channels: list[str] = field(default_factory=list) # Q13 + operational_challenge: str = "" # Q14 + biggest_pain_point: str = "" # Q15 + current_technology: list[str] = field(default_factory=list) # Q16 + marketing_channels: list[str] = field(default_factory=list) # Q17 + monthly_marketing_budget: float = 0.0 # Q18 MMK + + # ── Section 4: Goals & Constraints (6 questions) ───────────────────────── + goal_3_month: float = 0.0 # Q19 MMK + goal_6_month: float = 0.0 # Q20 MMK + goal_12_month: float = 0.0 # Q21 MMK + budget_constraint: str = "Moderate (200-500K)" # Q22 + tech_readiness: str = "Somewhat ready" # Q23 + preferred_language: str = "English" # Q24 + + +@dataclass +class HealthDimensions: + """Sub-scores for the five health dimensions (each 0–100).""" + revenue_strength: int = 0 + customer_retention: int = 0 + market_position: int = 0 + technology_adoption: int = 0 + growth_trajectory: int = 0 + + @property + def total(self) -> int: + """ + Official BIOS Health Score formula: + (Revenue Strength × 20) + (Customer Retention × 20) + + (Market Position × 20) + (Technology Adoption × 20) + + (Growth Trajectory × 20) + Each dimension is 0–100, weight is 20%, so max = 100. + """ + return round( + (self.revenue_strength * 0.20) + + (self.customer_retention * 0.20) + + (self.market_position * 0.20) + + (self.technology_adoption * 0.20) + + (self.growth_trajectory * 0.20) + ) + + def to_dict(self) -> dict: + return { + "revenue_strength": self.revenue_strength, + "customer_retention": self.customer_retention, + "market_position": self.market_position, + "technology_adoption": self.technology_adoption, + "growth_trajectory": self.growth_trajectory, + "total": self.total, + } + + +@dataclass +class Weakness: + rank: int + dimension: str + label: str + your_score: float + benchmark: float + gap: float + severity: str # HIGH / MEDIUM / LOW + detail: str + + def to_dict(self) -> dict: + return asdict(self) + + +@dataclass +class Opportunity: + rank: int + title: str + description: str + expected_impact: str + difficulty: str # EASY / MEDIUM / HARD + timeframe: str + revenue_uplift_mmk: Optional[float] = None + + def to_dict(self) -> dict: + return asdict(self) + + +@dataclass +class ActionItem: + priority: int + action: str + rationale: str + urgency_score: float + impact_score: float + feasibility_score: float + composite_score: float + + def to_dict(self) -> dict: + return asdict(self) + + +@dataclass +class DiagnosisReport: + """Full Module 1 output — the BIOS diagnosis report.""" + session_id: str + business_name: str + industry: str + location: str + generated_at: str + + health_score: int + health_label: str # Critical / Below Average / Fair / Good / Excellent + health_dimensions: HealthDimensions + + top_3_weaknesses: list[Weakness] + growth_opportunities: list[Opportunity] + priority_action_items: list[ActionItem] + + ai_narrative: str # BIOS LLM executive summary + benchmarking: list[dict] + next_module: str = "Strategy Engine (Module 2)" + + model_used: str = "" + generation_time_ms: int = 0 + + def to_dict(self) -> dict: + return { + "session_id": self.session_id, + "business_name": self.business_name, + "industry": self.industry, + "location": self.location, + "generated_at": self.generated_at, + "health_score": self.health_score, + "health_label": self.health_label, + "health_dimensions": self.health_dimensions.to_dict(), + "top_3_weaknesses": [w.to_dict() for w in self.top_3_weaknesses], + "growth_opportunities": [o.to_dict() for o in self.growth_opportunities], + "priority_action_items": [a.to_dict() for a in self.priority_action_items], + "ai_narrative": self.ai_narrative, + "benchmarking": self.benchmarking, + "next_module": self.next_module, + "model_used": self.model_used, + "generation_time_ms": self.generation_time_ms, + } + + def to_json(self, indent: int = 2) -> str: + return json.dumps(self.to_dict(), ensure_ascii=False, indent=indent) + + +# ═══════════════════════════════════════════════════════════════════════════════ +# MODEL ROUTER +# ═══════════════════════════════════════════════════════════════════════════════ + +class ModelRouter: + """ + Routes inference requests to the appropriate backend + model variant. + + Priority order when calling .infer(): + 1. If BIOS-Insight-v1 is flagged as available → use HF Inference API + 2. Else use base model via Groq (fastest, free tier) + 3. Fallback to Anthropic Claude + 4. Final fallback: MOCK mode (returns structured placeholder) + """ + + def __init__( + self, + backend: ModelBackend = ModelBackend.GROQ, + variant: ModelVariant = ModelVariant.BASE, + bios_insight_ready: bool = False, + temperature: float = 0.3, + max_tokens: int = 2048, + ): + self.backend = backend + self.variant = variant + self.bios_insight_ready = bios_insight_ready + self.temperature = temperature + self.max_tokens = max_tokens + + # Clients initialised lazily + self._groq_client: Any = None + self._hf_client: Any = None + self._anthropic_client: Any = None + + log.info( + f"ModelRouter initialised | backend={backend.value} " + f"variant={variant.value} | BIOS-Insight-v1 ready={bios_insight_ready}" + ) + + # ── Client factories ────────────────────────────────────────────────────── + + def _get_groq(self): + if self._groq_client is None: + if not GROQ_AVAILABLE: + raise RuntimeError("groq package not installed. Run: pip install groq") + api_key = os.getenv("GROQ_API_KEY") + if not api_key: + raise RuntimeError("GROQ_API_KEY environment variable not set") + self._groq_client = Groq(api_key=api_key) + return self._groq_client + + def _get_hf(self): + if self._hf_client is None: + if not HF_AVAILABLE: + raise RuntimeError("huggingface_hub not installed. Run: pip install huggingface_hub") + api_key = os.getenv("HF_API_KEY") + if not api_key: + raise RuntimeError("HF_API_KEY environment variable not set") + self._hf_client = InferenceClient(token=api_key) + return self._hf_client + + def _get_openai(self): + if self._openai_client is None: + if not OPENAI_AVAILABLE: + raise RuntimeError("openai package not installed. Run: pip install openai") + api_key = os.getenv("OPENAI_API_KEY") + if not api_key: + raise RuntimeError("OPENAI_API_KEY environment variable not set") + self._openai_client = openai.OpenAI(api_key=api_key) + return self._openai_client + + def _get_gemini(self): + if self._gemini_client is None: + if not GEMINI_AVAILABLE: + raise RuntimeError("google-generativeai package not installed. Run: pip install google-generativeai") + api_key = os.getenv("GEMINI_API_KEY") + if not api_key: + raise RuntimeError("GEMINI_API_KEY environment variable not set") + genai.configure(api_key=api_key) + self._gemini_client = genai.GenerativeModel('gemini-pro') + return self._gemini_client + + def _get_anthropic(self): + if self._anthropic_client is None: + if not ANTHROPIC_AVAILABLE: + raise RuntimeError("anthropic package not installed. Run: pip install anthropic") + api_key = os.getenv("ANTHROPIC_API_KEY") + if not api_key: + raise RuntimeError("ANTHROPIC_API_KEY not set") + self._anthropic_client = anthropic.Anthropic(api_key=api_key) + return self._anthropic_client + + # ── Routing decision ────────────────────────────────────────────────────── + + def _resolve_route(self) -> tuple[ModelBackend, ModelVariant]: + """Determine which backend + variant to actually use.""" + if self.bios_insight_ready and HF_AVAILABLE and os.getenv("HF_API_KEY"): + return ModelBackend.HF_INFERENCE, ModelVariant.BIOS_INSIGHT + if OPENAI_AVAILABLE and os.getenv("OPENAI_API_KEY"): + return ModelBackend.OPENAI, ModelVariant.BASE + if GEMINI_AVAILABLE and os.getenv("GEMINI_API_KEY"): + return ModelBackend.GEMINI, ModelVariant.BASE + if GROQ_AVAILABLE and os.getenv("GROQ_API_KEY"): + return ModelBackend.GROQ, ModelVariant.BASE + if ANTHROPIC_AVAILABLE and os.getenv("ANTHROPIC_API_KEY"): + return ModelBackend.ANTHROPIC, ModelVariant.BASE + return ModelBackend.MOCK, ModelVariant.BASE + + # ── Inference ───────────────────────────────────────────────────────────── + + def infer(self, system_prompt: str, user_prompt: str) -> tuple[str, str]: + """ + Send prompts to the resolved model. + + Returns: + (response_text, model_identifier_used) + """ + backend, variant = _resolve_route(self) if False else self._resolve_route() + model_id = GROQ_MODEL_IDS.get(variant, GROQ_MODEL_IDS[ModelVariant.BASE]) + log.info(f"Routing → backend={backend.value} model={model_id}") + + if backend == ModelBackend.GROQ: + return self._infer_groq(system_prompt, user_prompt, model_id) + + if backend == ModelBackend.HF_INFERENCE: + hf_model = MODEL_IDS[ModelVariant.BIOS_INSIGHT] + return self._infer_hf(system_prompt, user_prompt, hf_model) + + if backend == ModelBackend.OPENAI: + return self._infer_openai(system_prompt, user_prompt) + + if backend == ModelBackend.GEMINI: + return self._infer_gemini(system_prompt, user_prompt) + + if backend == ModelBackend.ANTHROPIC: + return self._infer_anthropic(system_prompt, user_prompt) + + # MOCK fallback + return self._mock_response(), "mock/bios-kernel-v1" + + def _infer_groq(self, system: str, user: str, model: str) -> tuple[str, str]: + client = self._get_groq() + response = client.chat.completions.create( + model=model, + messages=[ + {"role": "system", "content": system}, + {"role": "user", "content": user}, + ], + temperature=self.temperature, + max_tokens=self.max_tokens, + response_format={"type": "json_object"}, + ) + return response.choices[0].message.content, f"groq/{model}" + + def _infer_hf(self, system: str, user: str, model: str) -> tuple[str, str]: + client = self._get_hf() + messages = [ + {"role": "system", "content": system}, + {"role": "user", "content": user}, + ] + response = client.chat_completion( + messages=messages, + model=model, + max_tokens=self.max_tokens, + temperature=self.temperature, + ) + return response.choices[0].message.content, f"hf/{model}" + + def _infer_openai(self, system: str, user: str) -> tuple[str, str]: + client = self._get_openai() + response = client.chat.completions.create( + model="gpt-4", + messages=[ + {"role": "system", "content": system}, + {"role": "user", "content": user}, + ], + temperature=self.temperature, + max_tokens=self.max_tokens, + response_format={"type": "json_object"}, + ) + return response.choices[0].message.content, "openai/gpt-4" + + def _infer_gemini(self, system: str, user: str) -> tuple[str, str]: + client = self._get_gemini() + combined_prompt = f"{system}\n\n{user}" + response = client.generate_content( + combined_prompt, + generation_config={ + "temperature": self.temperature, + "max_output_tokens": self.max_tokens, + } + ) + return response.text, "gemini/gemini-pro" + + def _infer_anthropic(self, system: str, user: str) -> tuple[str, str]: + client = self._get_anthropic() + message = client.messages.create( + model="claude-sonnet-4-20250514", + max_tokens=self.max_tokens, + system=system, + messages=[{"role": "user", "content": user}], + ) + return message.content[0].text, "anthropic/claude-sonnet-4-20250514" + + def _mock_response(self) -> str: + """Return a valid JSON mock for offline testing.""" + return json.dumps({ + "narrative": ( + "BIOS analysis complete. Your business shows strong foundational " + "elements but faces challenges in customer retention and technology " + "adoption. Prioritise loyalty initiatives and digital tooling to " + "unlock the next growth tier." + ), + "model": "mock", + }) + + +# ════════���══════════════════════════════════════════════════════════════════════ +# DIAGNOSIS ENGINE (pure scoring logic — no LLM required) +# ═══════════════════════════════════════════════════════════════════════════════ + +class DiagnosisEngine: + """ + Implements the BIOS Module 1 scoring algorithms. + + All calculations are deterministic and reproducible — the LLM is only + used to generate the qualitative narrative on top of these numbers. + """ + + # ── Dimension scorers ───────────────────────────────────────────────────── + + @staticmethod + def score_revenue(monthly_revenue: float) -> int: + thresholds = [ + (50_000_000, 100), + (20_000_000, 80), + ( 5_000_000, 60), + ( 1_000_000, 40), + ] + for threshold, score in thresholds: + if monthly_revenue >= threshold: + return score + return 20 + + @staticmethod + def score_retention(rate: float) -> int: + thresholds = [(80, 100), (60, 80), (40, 60), (20, 40)] + for threshold, score in thresholds: + if rate >= threshold: + return score + return 20 + + @staticmethod + def score_market_position(usp: str, competitors: str) -> int: + words = len(usp.strip().split()) + base = 20 + if words >= 50: base = 80 + elif words >= 30: base = 60 + elif words >= 15: base = 40 + # Bonus for knowing your competition (+5, capped at 100) + if competitors and len(competitors.strip()) > 5: + base = min(100, base + 5) + return base + + @staticmethod + def score_technology(technology: list[str], industry: str = "Other") -> int: + tech_lower = [t.lower() for t in technology] + if not tech_lower or "none" in tech_lower: + return 10 + + # Technology startup specific scoring + if industry.lower() == "technology startup": + advanced = {"cloud services", "microservices", "ai/ml", "blockchain", "devops", "kubernetes", "docker", "serverless", "api integration", "saas platform", "mobile app", "web app", "database", "analytics", "automation", "ci/cd", "git", "agile"} + mid_tier = {"pos system", "accounting software", "inventory system", "crm", "erp", "project management", "collaboration tools"} + basic = {"spreadsheets", "facebook business suite", "whatsapp business", "email marketing"} + else: + # Traditional business scoring + advanced = {"erp", "crm", "ai tools", "automation", "bi dashboard"} + mid_tier = {"pos system", "accounting software", "inventory system"} + basic = {"spreadsheets", "facebook business suite", "whatsapp business"} + + if any(t in advanced for t in tech_lower): return 100 + if any(t in mid_tier for t in tech_lower): return 60 + if any(t in basic for t in tech_lower): return 30 + return 20 + + @staticmethod + def score_growth(goal_12: float, current: float) -> int: + if current <= 0: + return 40 # can't compute — neutral score + rate = (goal_12 - current) / current * 100 + thresholds = [(50, 100), (30, 80), (10, 60), (0, 40)] + for threshold, score in thresholds: + if rate >= threshold: + return score + return 20 + + # ── Main scorer ─────────────────────────────────────────────────────────── + + def compute_dimensions(self, inp: BusinessInputs) -> HealthDimensions: + return HealthDimensions( + revenue_strength = self.score_revenue(inp.monthly_revenue), + customer_retention = self.score_retention(inp.retention_rate), + market_position = self.score_market_position( + inp.unique_selling_proposition, + inp.main_competitors), + technology_adoption = self.score_technology(inp.current_technology, inp.industry), + growth_trajectory = self.score_growth( + inp.goal_12_month, + inp.monthly_revenue), + ) + + @staticmethod + def health_label(score: int) -> str: + if score >= 80: return "Excellent" + if score >= 65: return "Good" + if score >= 45: return "Fair" + if score >= 30: return "Below Average" + return "Critical" + + # ── Weakness identification ─────────────────────────────────────────────── + + def identify_weaknesses( + self, + inp: BusinessInputs, + dims: HealthDimensions, + ) -> list[Weakness]: + bench = INDUSTRY_BENCHMARKS.get(inp.industry, INDUSTRY_BENCHMARKS["Other"]) + bench_scores = { + "revenue_strength": self.score_revenue(bench["avg_revenue"]), + "customer_retention": self.score_retention(bench["avg_retention"]), + "market_position": 60, # industry-standard expectation + "technology_adoption": 60, + "growth_trajectory": 60, + } + labels = { + "revenue_strength": "Monthly Revenue", + "customer_retention": "Customer Retention", + "market_position": "Market Differentiation", + "technology_adoption": "Technology Adoption", + "growth_trajectory": "Growth Ambition", + } + details = { + "revenue_strength": f"Revenue of {inp.monthly_revenue:,.0f} MMK is significantly below the {inp.industry} industry average.", + "customer_retention": f"Only {inp.retention_rate:.0f}% repeat purchase rate — industry average is {bench['avg_retention']:.0f}%.", + "market_position": "Your unique selling proposition needs greater clarity and depth to stand out.", + "technology_adoption": "Low technology adoption is limiting operational efficiency and scalability.", + "growth_trajectory": "Growth goals are misaligned with current revenue trajectory.", + } + + user_scores = { + "revenue_strength": dims.revenue_strength, + "customer_retention": dims.customer_retention, + "market_position": dims.market_position, + "technology_adoption": dims.technology_adoption, + "growth_trajectory": dims.growth_trajectory, + } + + weaknesses: list[Weakness] = [] + for key, user_score in user_scores.items(): + b_score = bench_scores[key] + if user_score < b_score * 0.8: + gap = b_score - user_score + severity = "HIGH" if gap > 30 else "MEDIUM" if gap > 15 else "LOW" + weaknesses.append(Weakness( + rank=0, + dimension=key, + label=labels[key], + your_score=user_score, + benchmark=b_score, + gap=round(gap, 1), + severity=severity, + detail=details[key], + )) + + # Sort: HIGH first, then by gap descending + sev_order = {"HIGH": 3, "MEDIUM": 2, "LOW": 1} + weaknesses.sort(key=lambda w: (sev_order[w.severity], w.gap), reverse=True) + top3 = weaknesses[:3] + for i, w in enumerate(top3, 1): + w.rank = i + return top3 + + # ── Opportunity discovery ───────────────────────────────────────────────── + + def discover_opportunities( + self, + inp: BusinessInputs, + dims: HealthDimensions, + ) -> list[Opportunity]: + bench = INDUSTRY_BENCHMARKS.get(inp.industry, INDUSTRY_BENCHMARKS["Other"]) + opps: list[Opportunity] = [] + + # 1. Revenue growth + if inp.goal_12_month > inp.monthly_revenue and inp.monthly_revenue > 0: + pct = (inp.goal_12_month / inp.monthly_revenue - 1) * 100 + opps.append(Opportunity( + rank=0, + title="Scale Revenue Toward 12-Month Goal", + description=f"Bridge the {pct:.0f}% gap to your {inp.goal_12_month:,.0f} MMK annual revenue target.", + expected_impact=f"+{pct:.0f}% revenue growth", + difficulty="MEDIUM", + timeframe="6–12 months", + revenue_uplift_mmk=inp.goal_12_month - inp.monthly_revenue, + )) + + # 2. Retention improvement + if inp.retention_rate < bench["avg_retention"]: + gap = bench["avg_retention"] - inp.retention_rate + monthly_g = (gap / 100) * inp.avg_customer_lifetime_value * max(inp.team_size, 1) + opps.append(Opportunity( + rank=0, + title="Boost Customer Retention Rate", + description=( + f"Raise repeat-purchase rate by {gap:.0f}% to match the " + f"{inp.industry} industry benchmark." + ), + expected_impact=f"+{monthly_g:,.0f} MMK estimated monthly revenue", + difficulty="MEDIUM", + timeframe="2–3 months", + revenue_uplift_mmk=monthly_g * 12, + )) + + # 3. Channel expansion + if len(inp.sales_channels) < 3: + needed = 3 - len(inp.sales_channels) + opps.append(Opportunity( + rank=0, + title=f"Expand to {needed} New Sales Channel{'s' if needed > 1 else ''}", + description="Diversifying beyond your current channels reduces single-point risk and opens new customer pools.", + expected_impact="+20–30% customer reach", + difficulty="EASY", + timeframe="1–2 months", + )) + + # 4. Technology upgrade + if dims.technology_adoption < 60: + opps.append(Opportunity( + rank=0, + title="Adopt Core Business Technology", + description="Implementing a CRM or POS system unlocks data-driven decisions and staff efficiency.", + expected_impact="+15–25% operational efficiency", + difficulty="MEDIUM", + timeframe="2–4 weeks", + )) + + # 5. Marketing investment + if inp.monthly_marketing_budget < bench["avg_mkt"] * 0.5: + opps.append(Opportunity( + rank=0, + title="Increase Marketing Investment", + description=( + f"Current budget of {inp.monthly_marketing_budget:,.0f} MMK " + f"is far below the {inp.industry} average of {bench['avg_mkt']:,.0f} MMK." + ), + expected_impact="+10–20% new customer acquisition", + difficulty="EASY", + timeframe="1 month", + )) + + # Rank and cap at 5 + for i, opp in enumerate(opps[:5], 1): + opp.rank = i + return opps[:5] + + # ── Priority action items ───────────────────────────────────────────────── + + RECOMMENDED_ACTIONS: dict[str, str] = { + "revenue_strength": "Run a margin audit and introduce 2 high-value upsell products this month.", + "customer_retention": "Launch a loyalty stamp card and a 30-day follow-up WhatsApp message sequence.", + "market_position": "Rewrite your USP in one clear sentence and test it in Facebook ad copy.", + "technology_adoption": "Set up a free CRM (HubSpot or Zoho) and import your customer contact list.", + "growth_trajectory": "Break your 12-month target into monthly milestones and review weekly.", + } + + def rank_priority_actions( + self, + weaknesses: list[Weakness], + inp: BusinessInputs, + ) -> list[ActionItem]: + items: list[ActionItem] = [] + sev_urgency = {"HIGH": 85, "MEDIUM": 60, "LOW": 35} + + for w in weaknesses: + urgency = float(sev_urgency.get(w.severity, 40)) + impact = min(100.0, w.gap * 1.6) + feasibility = {"HIGH": 40.0, "MEDIUM": 65.0, "LOW": 80.0}.get(w.severity, 50.0) + composite = round(urgency * 0.4 + impact * 0.4 + feasibility * 0.2, 1) + + items.append(ActionItem( + priority=0, + action=self.RECOMMENDED_ACTIONS.get(w.dimension, f"Address {w.label} urgently."), + rationale=w.detail, + urgency_score=urgency, + impact_score=round(impact, 1), + feasibility_score=feasibility, + composite_score=composite, + )) + + items.sort(key=lambda x: x.composite_score, reverse=True) + for i, item in enumerate(items, 1): + item.priority = i + return items + + # ── Benchmarking table ──────────────────────────────────────────────────── + + def build_benchmarking( + self, + inp: BusinessInputs, + ) -> list[dict]: + bench = INDUSTRY_BENCHMARKS.get(inp.industry, INDUSTRY_BENCHMARKS["Other"]) + + def status(val: float, avg: float) -> str: + if val >= avg * 1.10: return "ABOVE" + if val <= avg * 0.90: return "BELOW" + return "AT" + + return [ + {"metric": "Monthly Revenue", "your_value": inp.monthly_revenue, "industry_avg": bench["avg_revenue"], "unit": "MMK", "status": status(inp.monthly_revenue, bench["avg_revenue"])}, + {"metric": "Customer Retention Rate", "your_value": inp.retention_rate, "industry_avg": bench["avg_retention"], "unit": "%", "status": status(inp.retention_rate, bench["avg_retention"])}, + {"metric": "Avg Customer Lifetime Val","your_value": inp.avg_customer_lifetime_value, "industry_avg": bench["avg_clv"], "unit": "MMK", "status": status(inp.avg_customer_lifetime_value, bench["avg_clv"])}, + {"metric": "Marketing Budget", "your_value": inp.monthly_marketing_budget, "industry_avg": bench["avg_mkt"], "unit": "MMK", "status": status(inp.monthly_marketing_budget, bench["avg_mkt"])}, + {"metric": "Team Size", "your_value": float(inp.team_size), "industry_avg": float(bench["avg_team"]),"unit": "ppl", "status": status(inp.team_size, bench["avg_team"])}, + ] + + +# ═══════════════════════════════════════════════════════════════════════════════ +# INSIGHT GENERATOR (builds the LLM prompt and parses the response) +# ═══════════════════════════════════════════════════════════════════════════════ + +class InsightGenerator: + """ + Constructs structured prompts for the BIOS LLM and parses its JSON output + into the qualitative `ai_narrative` field of the DiagnosisReport. + """ + + SYSTEM_PROMPT = """You are BIOS — the Business Idea Operating System. You are the elite AI advisor for Myanmar SMEs, Gold Shops, and ambitious entrepreneurs across Southeast Asia. + +Your personality: professional, precise, encouraging, and bold — like a McKinsey partner who speaks to founders, not just analysts. You use the Dark & Gold luxury tone: every word carries weight, every recommendation is actionable. + +You always respond in valid JSON with this exact structure: +{ + "narrative": "<3-paragraph executive summary in the user's preferred language>", + "headline_insight": "" +} + +Rules: +- Be specific with numbers from the data provided +- Use the language specified in preferred_language +- Never be generic — reference the actual business, industry, and goals +- Tone: elite advisory, not chatbot small talk""" + + def build_user_prompt(self, inp: BusinessInputs, dims: HealthDimensions, weaknesses: list[Weakness], opps: list[Opportunity]) -> str: + weak_lines = "\n".join( + f" {w.rank}. {w.label} — score {w.your_score:.0f} vs benchmark {w.benchmark:.0f} | severity: {w.severity}" + for w in weaknesses + ) + opp_lines = "\n".join( + f" {o.rank}. {o.title}: {o.expected_impact} ({o.timeframe})" + for o in opps[:3] + ) + + return f"""BIOS DIAGNOSIS DATA — analyse and generate insights. + +Business: {inp.business_name} +Industry: {inp.industry} | Location: {inp.location} +Years Operating: {inp.years_in_business} | Team: {inp.team_size} people +Monthly Revenue: {inp.monthly_revenue:,.0f} MMK +Retention Rate: {inp.retention_rate:.0f}% +Monthly Marketing Budget: {inp.monthly_marketing_budget:,.0f} MMK +Preferred Language: {inp.preferred_language} + +HEALTH SCORE: {dims.total}/100 — {DiagnosisEngine.health_label(dims.total)} + Revenue Strength: {dims.revenue_strength} + Customer Retention: {dims.customer_retention} + Market Position: {dims.market_position} + Technology Adoption: {dims.technology_adoption} + Growth Trajectory: {dims.growth_trajectory} + +TOP WEAKNESSES: +{weak_lines} + +TOP OPPORTUNITIES: +{opp_lines} + +12-Month Revenue Goal: {inp.goal_12_month:,.0f} MMK +Unique Value Proposition: "{inp.unique_selling_proposition}" +Biggest Pain Point: "{inp.biggest_pain_point}" + +Generate the executive narrative JSON now.""" + + def parse_narrative(self, raw: str) -> str: + """Extract narrative from LLM JSON response, with fallback.""" + try: + # Strip markdown fences if present + clean = re.sub(r"```(?:json)?\s*", "", raw).strip().rstrip("```").strip() + data = json.loads(clean) + return data.get("narrative", raw) + except (json.JSONDecodeError, AttributeError): + # Return raw text if not parseable + return raw.strip() + + +# ═══════════════════════════════════════════════════════════════════════════════ +# NEON DB WRITER +# ═══════════════════════════════════════════════════════════════════════════════ + +class NeonDBWriter: + """ + Persists BIOS diagnosis reports to NeonDB (PostgreSQL) via psycopg v3. + + Required table (run schema_auth.sql first): + CREATE TABLE IF NOT EXISTS diagnoses ( + id UUID PRIMARY KEY DEFAULT uuid_generate_v4(), + session_id VARCHAR(255) UNIQUE NOT NULL, + business_name VARCHAR(255), + industry VARCHAR(100), + location VARCHAR(100), + health_score INTEGER, + health_label VARCHAR(50), + health_dimensions JSONB, + top_3_weaknesses JSONB, + growth_opportunities JSONB, + priority_action_items JSONB, + ai_narrative TEXT, + benchmarking JSONB, + model_used VARCHAR(255), + generation_time_ms INTEGER, + status VARCHAR(50) DEFAULT 'COMPLETED', + created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() + ); + """ + + def __init__(self, database_url: Optional[str] = None): + self.database_url = database_url or os.getenv("DATABASE_URL") + if not self.database_url: + raise ValueError( + "DATABASE_URL not set. Export it or pass database_url= to NeonDBWriter." + ) + # Normalise asyncpg URL to psycopg URL + self.database_url = self.database_url.replace( + "postgresql+asyncpg://", "postgresql://" + ).replace( + "postgres+asyncpg://", "postgresql://" + ) + + def save_report(self, report: DiagnosisReport) -> str: + """ + Upsert a DiagnosisReport into the diagnoses table. + Returns the session_id of the saved record. + """ + d = report.to_dict() + + sql = """ + INSERT INTO diagnoses ( + session_id, business_name, industry, location, + health_score, health_label, health_dimensions, + top_3_weaknesses, growth_opportunities, priority_action_items, + ai_narrative, benchmarking, model_used, generation_time_ms, + status, created_at + ) VALUES ( + %(session_id)s, %(business_name)s, %(industry)s, %(location)s, + %(health_score)s, %(health_label)s, %(health_dimensions)s, + %(top_3_weaknesses)s, %(growth_opportunities)s, %(priority_action_items)s, + %(ai_narrative)s, %(benchmarking)s, %(model_used)s, %(generation_time_ms)s, + 'COMPLETED', NOW() + ) + ON CONFLICT (session_id) DO UPDATE SET + health_score = EXCLUDED.health_score, + health_label = EXCLUDED.health_label, + health_dimensions = EXCLUDED.health_dimensions, + top_3_weaknesses = EXCLUDED.top_3_weaknesses, + growth_opportunities = EXCLUDED.growth_opportunities, + priority_action_items = EXCLUDED.priority_action_items, + ai_narrative = EXCLUDED.ai_narrative, + benchmarking = EXCLUDED.benchmarking, + model_used = EXCLUDED.model_used, + generation_time_ms = EXCLUDED.generation_time_ms, + status = 'COMPLETED' + """ + + params = { + "session_id": d["session_id"], + "business_name": d["business_name"], + "industry": d["industry"], + "location": d["location"], + "health_score": d["health_score"], + "health_label": d["health_label"], + "health_dimensions": json.dumps(d["health_dimensions"]), + "top_3_weaknesses": json.dumps(d["top_3_weaknesses"]), + "growth_opportunities": json.dumps(d["growth_opportunities"]), + "priority_action_items": json.dumps(d["priority_action_items"]), + "ai_narrative": d["ai_narrative"], + "benchmarking": json.dumps(d["benchmarking"]), + "model_used": d["model_used"], + "generation_time_ms": d["generation_time_ms"], + } + + with psycopg.connect(self.database_url) as conn: + with conn.cursor() as cur: + cur.execute(sql, params) + conn.commit() + + log.info(f"✅ Report saved to NeonDB | session_id={report.session_id} | score={report.health_score}") + return report.session_id + + def fetch_report(self, session_id: str) -> Optional[dict]: + """Fetch a previously saved report by session_id.""" + sql = "SELECT * FROM diagnoses WHERE session_id = %s" + with psycopg.connect(self.database_url, row_factory=dict_row) as conn: + with conn.cursor() as cur: + cur.execute(sql, (session_id,)) + row = cur.fetchone() + return dict(row) if row else None + + def list_reports(self, limit: int = 20) -> list[dict]: + """Return the most recent diagnoses.""" + sql = "SELECT session_id, business_name, industry, health_score, health_label, status, created_at FROM diagnoses ORDER BY created_at DESC LIMIT %s" + with psycopg.connect(self.database_url, row_factory=dict_row) as conn: + with conn.cursor() as cur: + cur.execute(sql, (limit,)) + rows = cur.fetchall() + return [dict(r) for r in rows] + + +# ═══════════════════════════════════════════════════════════════════════════════ +# BIOS CONTROLLER — the main entry point +# ═══════════════════════════════════════════════════════════════════════════════ + +class BIOSController: + """ + BIOS-kernel-v1 · Business Idea Operating System · Module 1 Controller + + Orchestrates the full diagnosis pipeline: + inputs → scoring → LLM narrative → structured report → NeonDB + + Usage: + controller = BIOSController() + report = controller.run_diagnosis(inputs) + print(report.to_json()) + + To use the fine-tuned BIOS-Insight-v1 when it becomes available on HF: + controller = BIOSController(bios_insight_ready=True) + """ + + VERSION = "1.0.0" + KERNEL = "BIOS-kernel-v1" + + def __init__( + self, + backend: ModelBackend = ModelBackend.GROQ, + bios_insight_ready: bool = False, + temperature: float = 0.3, + max_tokens: int = 2048, + database_url: Optional[str] = None, + save_to_db: bool = True, + ): + self.router = ModelRouter( + backend=backend, + bios_insight_ready=bios_insight_ready, + temperature=temperature, + max_tokens=max_tokens, + ) + self.engine = DiagnosisEngine() + self.generator = InsightGenerator() + self.save_to_db = save_to_db + + # DB writer — lazy init so missing DB URL doesn't break non-DB usage + self._db: Optional[NeonDBWriter] = None + self._db_url = database_url + + log.info( + f"╔═══════════════════════════════════════════╗\n" + f" BIOS Controller v{self.VERSION} · {self.KERNEL}\n" + f" backend={backend.value} | save_db={save_to_db}\n" + f"╚═══════════════════════════════════════════╝" + ) + + @property + def db(self) -> NeonDBWriter: + if self._db is None: + self._db = NeonDBWriter(self._db_url) + return self._db + + # ── Main pipeline ───────────────────────────────────────────────────────── + + def run_diagnosis(self, inputs: BusinessInputs) -> DiagnosisReport: + """ + Full Module 1 pipeline. + + Args: + inputs: Completed BusinessInputs with all 24 question answers. + + Returns: + DiagnosisReport with health_score, top_3_weaknesses, + growth_opportunities, priority_action_items, and ai_narrative. + """ + t_start = time.perf_counter() + session_id = str(uuid.uuid4()) + + log.info(f"▶ Starting BIOS diagnosis | business='{inputs.business_name}' | session={session_id}") + + # ── Step 1: Compute health dimensions and score ────────────────────── + dims = self.engine.compute_dimensions(inputs) + score = dims.total + label = self.engine.health_label(score) + log.info(f" Health Score: {score}/100 ({label})") + log.info(f" Dimensions: Rev={dims.revenue_strength} Ret={dims.customer_retention} Mkt={dims.market_position} Tech={dims.technology_adoption} Grow={dims.growth_trajectory}") + + # ── Step 2: Identify weaknesses ────────────────────────────────────── + weaknesses = self.engine.identify_weaknesses(inputs, dims) + log.info(f" Weaknesses identified: {[w.label for w in weaknesses]}") + + # ── Step 3: Discover opportunities ─────────────────────────────────── + opportunities = self.engine.discover_opportunities(inputs, dims) + log.info(f" Opportunities found: {len(opportunities)}") + + # ── Step 4: Rank priority actions ──────────────────────────────────── + actions = self.engine.rank_priority_actions(weaknesses, inputs) + + # ── Step 5: Build benchmarking table ───────────────────────────────── + benchmarking = self.engine.build_benchmarking(inputs) + + # ── Step 6: Generate AI narrative via LLM ──────────────────────────── + narrative, model_used = self._generate_narrative(inputs, dims, weaknesses, opportunities) + log.info(f" Narrative generated | model={model_used}") + + t_ms = int((time.perf_counter() - t_start) * 1000) + + # ── Step 7: Assemble report ─────────────────────────────────────────── + report = DiagnosisReport( + session_id = session_id, + business_name = inputs.business_name, + industry = inputs.industry, + location = inputs.location, + generated_at = datetime.now(timezone.utc).isoformat(), + health_score = score, + health_label = label, + health_dimensions = dims, + top_3_weaknesses = weaknesses, + growth_opportunities = opportunities, + priority_action_items= actions, + ai_narrative = narrative, + benchmarking = benchmarking, + model_used = model_used, + generation_time_ms = t_ms, + ) + + log.info(f"✔ Diagnosis complete | score={score} | {t_ms}ms") + + # ── Step 8: Save to NeonDB ──────────────────────────────────────────── + if self.save_to_db: + try: + self.db.save_report(report) + except Exception as e: + log.warning(f"DB save failed (non-fatal): {e}") + + return report + + def _generate_narrative( + self, + inp: BusinessInputs, + dims: HealthDimensions, + weak: list[Weakness], + opps: list[Opportunity], + ) -> tuple[str, str]: + """Call the LLM and return (narrative_text, model_identifier).""" + system = self.generator.SYSTEM_PROMPT + user = self.generator.build_user_prompt(inp, dims, weak, opps) + try: + raw, model_id = self.router.infer(system, user) + narrative = self.generator.parse_narrative(raw) + return narrative, model_id + except Exception as e: + log.warning(f"LLM call failed, using fallback narrative: {e}") + fallback = ( + f"{inp.business_name} received a BIOS Health Score of {dims.total}/100 ({self.engine.health_label(dims.total)}). " + f"Key areas for immediate attention: {', '.join(w.label for w in weak[:2])}. " + f"Top opportunity: {opps[0].title if opps else 'revenue diversification'}." + ) + return fallback, "fallback/static" + + # ── Convenience helpers ─────────────────────────────────────────────────── + + def switch_to_bios_insight(self): + """Activate BIOS-Insight-v1 once it is published on HuggingFace.""" + self.router.bios_insight_ready = True + self.router.variant = ModelVariant.BIOS_INSIGHT + log.info("🌟 Switched to BIOS-Insight-v1 (fine-tuned model)") + + def switch_to_base(self): + """Revert to base llama-3.3-70b model.""" + self.router.bios_insight_ready = False + self.router.variant = ModelVariant.BASE + log.info("Reverted to base model (llama-3.3-70b)") + + def get_report(self, session_id: str) -> Optional[dict]: + """Retrieve a saved report from NeonDB.""" + return self.db.fetch_report(session_id) + + def list_reports(self, limit: int = 20) -> list[dict]: + """List recent diagnosis reports from NeonDB.""" + return self.db.list_reports(limit) + + +# ═══════════════════════════════════════════════════════════════════════════════ +# CLI / DEMO RUNNER +# ═══════════════════════════════════════════════════════════════════════════════ + +def _demo_inputs() -> BusinessInputs: + """Sample Technology Startup business for demonstration.""" + return BusinessInputs( + # Section 1 + business_name = "MyanmarTech Solutions", + industry = "Technology Startup", + location = "Yangon", + years_in_business = 2, + monthly_revenue = 8_500_000, + team_size = 12, + # Section 2 + target_customer = "SMEs in Myanmar seeking digital transformation and automation solutions.", + acquisition_channels = ["LinkedIn", "Tech Meetups", "Referrals", "Online Ads"], + avg_customer_lifetime_value = 3_200_000, + retention_rate = 75.0, + main_competitors = "Digital Myanmar, TechHub Asia, CloudBase Solutions", + unique_selling_proposition = "We provide AI-powered business automation specifically designed for Myanmar SMEs, with local language support and compliance.", + # Section 3 + sales_channels = ["SaaS Platform", "Direct Sales", "Partners"], + operational_challenge = "Scaling infrastructure while maintaining service quality", + biggest_pain_point = "Customer onboarding complexity - need streamlined setup process", + current_technology = ["Cloud Services", "AI/ML", "Microservices", "DevOps", "CI/CD"], + marketing_channels = ["LinkedIn", "Google Ads", "Content Marketing", "Tech Events"], + monthly_marketing_budget= 1_200_000, + # Section 4 + goal_3_month = 12_000_000, + goal_6_month = 18_000_000, + goal_12_month = 35_000_000, + budget_constraint = "Flexible (1M+)", + tech_readiness = "Very ready", + preferred_language = "English", + ) + + +if __name__ == "__main__": + print("\n" + "═" * 60) + print(" BIOS — Business Idea Operating System") + print(" BIOS-kernel-v1 · Module 1: Business Diagnosis") + print("═" * 60 + "\n") + + # ── Instantiate controller ──────────────────────────────────────────────── + # Set save_to_db=True and export DATABASE_URL to persist to NeonDB. + controller = BIOSController( + backend = ModelBackend.GROQ, + save_to_db = bool(os.getenv("DATABASE_URL")), + ) + + # ── Run diagnosis ───────────────────────────────────────────────────────── + inputs = _demo_inputs() + report = controller.run_diagnosis(inputs) + + # ── Print structured JSON output ────────────────────────────────────────── + print("\n" + "─" * 60) + print(" BIOS DIAGNOSIS REPORT") + print("─" * 60) + print(report.to_json()) + + print("\n" + "═" * 60) + print(f" Health Score : {report.health_score}/100 ({report.health_label})") + print(f" Session ID : {report.session_id}") + print(f" Model Used : {report.model_used}") + print(f" Generated in : {report.generation_time_ms}ms") + print("═" * 60 + "\n")