Merge branch 'online-st'

clients/groq_client.py

@@ -176,6 +176,8 @@ class GroqClient:
 
     # ---------------MAIN-----------------
 
+# Non-streaming ask method for backwards compatibility
+# ============================================
     def ask(self, raw_query: str) -> str:
         if (
             not raw_query
@@ -255,3 +257,91 @@ class GroqClient:
 
         # If all models fail
         return "I'm temporarily experiencing high demand. Please try again in a few minutes or rephrase your question."
+
+# Streaming ask method (Generator)
+# ============================================
+    def ask_stream(self, raw_query: str):
+        """Generator function that yields response chunks in real-time"""
+
+        # Validation
+        if not raw_query or raw_query is None or raw_query == "" or len(raw_query) > 1000:
+            yield "Please provide a valid query under 1,000 characters."
+            return
+
+        if raw_query.lower() in GREETINGS_TRIGGERS:
+            yield random.choice(GREETINGS)
+            return
+
+        # Retrieve documents
+        try:
+            docs = self.retriever.invoke(raw_query)
+        except Exception as e:
+            yield f"Error retrieving documents: {e}"
+            return
+
+        if not docs:
+            yield self.handle_unknown_query()
+            return
+
+        # Prepare context and prompt
+        context = "\n".join([d.page_content for d in docs])
+        fallback = self.handle_unknown_query()
+        prompt = self.PROMPT_TEMPLATE.format(
+            context=context, question=raw_query, fallback_response=fallback
+        )
+
+        messages = [
+            {
+                "role": "system",
+                "content": self.SYSTEM_MESSSAGE,
+            },
+            {"role": "user", "content": prompt},
+        ]
+
+        # Try multiple models with fallback
+        models_to_try = [
+            "openai/gpt-oss-120b",
+            "openai/gpt-oss-20b",
+            "compound-beta-mini",
+            "llama-3.1-8b-instant",
+            "llama-3.3-70b-versatile",
+        ]
+
+        random.shuffle(models_to_try)
+
+        for model in models_to_try:
+            try:
+                completion = self.client.chat.completions.create(
+                    model=model,
+                    messages=messages,
+                    temperature=TEMPERATURE,
+                    max_completion_tokens=MAX_TOKENS,
+                    top_p=TOP_P,
+                    stream=True,  # Enable streaming
+                )
+
+                # Stream chunks as they arrive
+                has_content = False
+                for chunk in completion:
+                    if chunk.choices[0].delta.content:
+                        has_content = True
+                        yield chunk.choices[0].delta.content
+
+                    # Check if stream finished
+                    if chunk.choices[0].finish_reason:
+                        break
+
+                # If we got content, we're done
+                if has_content:
+                    return
+
+            except Exception as e:
+                # Check if it's a rate limit error
+                if "rate_limit_exceeded" in str(e) or "429" in str(e):
+                    continue  # Try next model
+                else:
+                    yield f"Error while calling LLM: {e}"
+                    return
+
+        # If all models fail
+        yield "I'm temporarily experiencing high demand. Please try again in a few minutes or rephrase your question."

main.py

@@ -3,6 +3,7 @@ from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel
 from functools import lru_cache
 from clients.groq_client import GroqClient
+from fastapi.responses import StreamingResponse
 
 app = FastAPI()
 
@@ -12,7 +13,7 @@ origins = [
     "http://localhost:4200",
     "https://mosesmangabo.vercel.app",
     "https://chatbot-bridge-service.onrender.com",
-    "https://eng-musa-chatbotbridge.hf.space"
+    "https://eng-musa-chatbotbridge.hf.space",
 ]
 
 app.add_middleware(
@@ -28,11 +29,15 @@ class QueryRequest(BaseModel):
     query: str
     ip: str
 
+
 # Lazy singleton for GroqClient
 @lru_cache(maxsize=1)
 def get_groq_client():
     return GroqClient()
 
+
+# Non - Streaming
+# ============================================
 @app.post("/chat")
 async def chat_endpoint(request: QueryRequest):
     try:
@@ -46,10 +51,49 @@ async def chat_endpoint(request: QueryRequest):
             "next_questions": groq_client.get_next_questions(request.ip),
         }
     except Exception as e:
-        return {"status_code": 500, "response": f"Internal error: {e}", "next_questions": []}
+        return {
+            "status_code": 500,
+            "response": f"Internal error: {e}",
+            "next_questions": [],
+        }
+
+
+# Streaming
+# ============================================
+@app.post("/chat-stream")
+async def chat_stream_endpoint(request: QueryRequest):
+    """Streaming endpoint - returns chunks in real-time"""
+    try:
+        groq_client = get_groq_client()
+
+        def generate():
+            try:
+                for chunk in groq_client.ask_stream(request.query):
+                    yield f"data: {chunk}\n\n"
+
+                # Send completion signal
+                yield "data: [DONE]\n\n"
+
+            except Exception as e:
+                yield f"data: Error: {str(e)}\n\n"
+
+        return StreamingResponse(
+            generate(),
+            media_type="text/event-stream",
+            headers={
+                "Cache-Control": "no-cache",
+                "Connection": "keep-alive",
+                "X-Accel-Buffering": "no",  # Disable nginx buffering
+            },
+        )
+
+    except Exception as e:
+        return {"status_code": 500, "response": f"Internal error: {e}"}
+
 
 @app.get("/")
 def healthcheck():
     return {"status": "ok"}
 
-#python -m uvicorn main:app --reload
+
+# python -m uvicorn main:app --reload