Spaces:
Running
Running
| import json | |
| from fastapi import APIRouter, HTTPException, Depends | |
| from fastapi.responses import StreamingResponse | |
| from pydantic import BaseModel | |
| from firebase_admin import firestore | |
| from utils.rate_limiter import with_rate_limit | |
| from utils.groq_client import call_groq_stream, LIGHT_MODEL, HEAVY_MODEL | |
| from utils.pinecone_client import query_book_context | |
| from utils import business_type | |
| router = APIRouter() | |
| # ─── Schemas ────────────────────────────────────────────────────────────────── | |
| class Message(BaseModel): | |
| role: str # "user" or "assistant" | |
| content: str | |
| class ChatRequest(BaseModel): | |
| businessId: str | |
| sessionId: str | |
| message: str | |
| history: list[Message] = [] | |
| useHeavyModel: bool = False # True for AI Consultation sessions | |
| # ─── Helpers ────────────────────────────────────────────────────────────────── | |
| HISTORY_LIMIT = 10 # last N messages kept for context window efficiency | |
| HISTORY_CONTENT_CHARS = 4000 # per-message cap so one huge paste can't bloat every call | |
| def build_messages(system_prompt: str, history: list[Message], message: str) -> list[dict]: | |
| """ | |
| System prompt + recent history + the new message. Only 'user' and | |
| 'assistant' roles are accepted from the client: history comes from the | |
| browser, so a 'system' role in it would let a caller inject instructions. | |
| """ | |
| messages = [{"role": "system", "content": system_prompt}] | |
| for msg in history[-HISTORY_LIMIT:]: | |
| if msg.role in ("user", "assistant"): | |
| messages.append({"role": msg.role, "content": msg.content[:HISTORY_CONTENT_CHARS]}) | |
| messages.append({"role": "user", "content": message}) | |
| return messages | |
| def _get_business_context(user_id: str, business_id: str) -> str: | |
| """Builds a system context string from BFF + profile_output in Firestore.""" | |
| db = firestore.client() | |
| biz_ref = ( | |
| db.collection("user_data") | |
| .document(user_id) | |
| .collection("businesses") | |
| .document(business_id) | |
| ) | |
| # Fetch BFF section 1 (WHAT) for quick context | |
| s1 = biz_ref.collection("bff").document("section1").get() | |
| # Fetch business identity if generated | |
| identity = biz_ref.collection("profile_output").document("businessIdentity").get() | |
| ctx_parts = [] | |
| if s1.exists: | |
| d = s1.to_dict() | |
| t = business_type.describe(d) | |
| ctx_parts.append( | |
| f"Business: {d.get('businessName', 'Unknown')}, " | |
| f"Sector: {t['sector']}, " | |
| f"Type of business: {t['subsector']}, " | |
| f"Niche: {t['niche']}" | |
| ) | |
| if identity.exists: | |
| d = identity.to_dict() | |
| ctx_parts.append(f"Summary: {d.get('summary', '')}") | |
| ctx_parts.append(f"Value Proposition: {d.get('valueProposition', '')}") | |
| return "\n".join(ctx_parts) if ctx_parts else "Business profile not yet generated." | |
| # ─── Route: Chat message (streaming SSE) ───────────────────────────────────── | |
| def chat_message( | |
| body: ChatRequest, | |
| user: dict = Depends(with_rate_limit("chat")), | |
| ): | |
| """ | |
| Accepts a user message and returns a streaming SSE response. | |
| For passive AI Chat: uses light model (LIGHT_MODEL). | |
| For AI Consultation: uses heavy model (HEAVY_MODEL). | |
| Stream format (SSE): | |
| data: {"type": "thinking", "token": "..."} ← reasoning (heavy model only) | |
| data: {"type": "response", "token": "..."} ← main response | |
| data: {"type": "done"} ← stream end | |
| """ | |
| user_id = user.get("uid") | |
| model = HEAVY_MODEL if body.useHeavyModel else LIGHT_MODEL | |
| business_context = _get_business_context(user_id, body.businessId) | |
| # Retrieve relevant passages from "Built on Rock" based on user's message | |
| book_ctx = query_book_context(body.message, top_k=2) | |
| system_prompt = f"""You are the VenThrive Business Assistant — a business consultant trained in the Business Footing Framework (BFF) from "Built on Rock" by Iyanuoluwa Joseph Akinnusi. You help Nigerian entrepreneurs build successful businesses using the WHAT-WHY-HOW philosophy. | |
| BUSINESS CONTEXT: | |
| {business_context} | |
| {f"{chr(10)}RELEVANT BFF GUIDANCE FROM 'BUILT ON ROCK':{chr(10)}{book_ctx}" if book_ctx else ""} | |
| RULES: | |
| 1. STRICTLY BUSINESS: Stay focused on the entrepreneur's business. If asked personal questions, redirect warmly. | |
| 2. NIGERIAN CONTEXT: Reference real Nigerian examples — banks (GTBank, First Bank, Moniepoint), platforms (Jumia, Konga, Flutterwave), markets (Balogun, Alaba), cities and states. | |
| 3. BFF PHILOSOPHY: When relevant, ground advice in the WHAT-WHY-HOW framework from "Built on Rock". | |
| 4. PLAIN ENGLISH: No business jargon without explanation. Write like you're talking to a smart friend. | |
| 5. CONCISE: 5-6 sentences max per response. Use bullet points for lists. | |
| 5. ACTIONABLE: Every response should tell the entrepreneur what to DO next, not just what things mean. | |
| """ | |
| messages = build_messages(system_prompt, body.history, body.message) | |
| def stream(): | |
| try: | |
| # call_groq_stream handles key rotation on 429 | |
| stream_response = call_groq_stream( | |
| system_prompt=system_prompt, | |
| user_prompt=body.message, | |
| model=model, | |
| max_tokens=512, | |
| fixture="chat_response", | |
| messages=messages, | |
| ) | |
| for chunk in stream_response: | |
| token = chunk.choices[0].delta.content | |
| if token: | |
| yield f"data: {json.dumps({'type': 'response', 'token': token})}\n\n" | |
| yield f"data: {json.dumps({'type': 'done'})}\n\n" | |
| except Exception as e: | |
| yield f"data: {json.dumps({'type': 'error', 'message': 'Something went wrong. Please try again.'})}\n\n" | |
| return StreamingResponse(stream(), media_type="text/event-stream") | |