Stream LLM Responses (#18)

* feat: add SSE streaming endpoint for faster response display

Add /chat-stream SSE endpoint that sends advisor responses incrementally as they complete. Rewrite ChatPage handleSendMessage to consume SSE events. Add synthesize_responses method to orchestrator for aggregate mode support.

Made-with: Cursor

* Clean up changes to reduce diff and clean up unused imports

* Replace removed comments for code clarity

* Remove vestigial synthesized response handling

* Remove invalid kwarg

* Implement Cursor-generated fix to parallel response generation

* Undo `json` import name override

* Clean up unused params

* Refactor repeated message handling to match existing implementation

* Refactor to return JSON objects instead of specifically-formatted plaintext

* Remove trailing newline to reduce PR diff

* Update `chat_sequential_enhanced` docstring to be consistent with streaming response docstring

* Address review comments

* Remove unused `active_advisors` handling to simplify advisor selection

---------

Co-authored-by: NeonClary <askclary@gmail.com>

multi_llm_chatbot_backend/app/api/routes/chat.py

@@ -1,15 +1,20 @@
-from fastapi import APIRouter, Request, HTTPException, Body, Depends
-from app.models.persona import Persona
-from app.core.session_manager import get_session_manager
+import asyncio
+import json
+import logging
+import traceback
+from typing import Any, Dict, List, Literal, Optional
+
+from bson import ObjectId
+from fastapi import APIRouter, Depends, HTTPException, Request
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel, Field
+
 from app.api.utils import get_or_create_session_for_request_async
-from app.core.bootstrap import chat_orchestrator
 from app.core.auth import get_current_active_user
-from app.models.user import User
-from pydantic import BaseModel
-from typing import Optional
-import logging
+from app.core.bootstrap import chat_orchestrator
 from app.core.database import get_database
-from bson import ObjectId
+from app.core.session_manager import get_session_manager
+from app.models.user import User
 
 logger = logging.getLogger(__name__)
 
@@ -25,6 +30,7 @@ class ChatMessage(BaseModel):
     session_id: Optional[str] = None
     chat_session_id: Optional[str] = None  # MongoDB chat session ID
     response_length: str = "medium"
+    active_advisors: Optional[List[str]] = None
 
 class ReplyToAdvisor(BaseModel):
     user_input: str
@@ -42,6 +48,140 @@ class SwitchChatRequest(BaseModel):
 class NewChatRequest(BaseModel):
     title: Optional[str] = "New Chat"
 
+ChatStreamEventType = Literal["error", "progress", "clarification", "advisor"]
+
+
+class ChatStreamLine(BaseModel):
+    """One NDJSON line from ``/chat-stream``."""
+
+    type: ChatStreamEventType
+    data: Dict[str, Any] = Field(default_factory=dict)
+
+    def to_ndjson(self) -> str:
+        return json.dumps(self.model_dump(mode="json"), ensure_ascii=False) + "\n"
+
+
+@router.post("/chat-stream")
+async def chat_stream(
+    message: ChatMessage,
+    request: Request,
+    current_user: User = Depends(get_current_active_user),
+) -> StreamingResponse:
+    """
+    Streaming variant of chat-sequential (newline-delimited JSON).
+    @param message: ChatMessage containing user input and optional session/chat IDs
+    @param request: FastAPI Request object for session management
+    @param current_user: Authenticated user from dependency injection
+    @return: StreamingResponse that yields ChatStreamLine events as NDJSON
+    """
+
+    async def _event_generator():
+        try:
+            if message.chat_session_id:
+                sid = f"chat_{message.chat_session_id}"
+                if sid not in session_manager.sessions:
+                    sid = await get_or_create_session_for_request_async(
+                        request,
+                        chat_session_id=message.chat_session_id,
+                        user_id=str(current_user.id),
+                    )
+            else:
+                sid = await get_or_create_session_for_request_async(request)
+
+            session = session_manager.get_session(sid)
+
+            if (
+                len(session.messages) > 0 and
+                session.messages[-1].get("role") == "user" and
+                session.messages[-1].get("content") == message.user_input
+            ):
+                # TODO: This should be handled in the front-end input
+                logger.warning(f"Repeated user input: {message.user_input}")
+
+            session.append_message("user", message.user_input)
+
+            if chat_orchestrator.needs_clarification(session, message.user_input):
+                clar = await chat_orchestrator.generate_contextual_clarification(message.user_input)
+                yield ChatStreamLine(
+                    type="clarification",
+                    data={
+                        "message": clar["question"],
+                        "suggestions": clar["suggestions"],
+                    },
+                ).to_ndjson()
+                yield ChatStreamLine(
+                    type="progress",
+                    data={"phase": "complete"},
+                ).to_ndjson()
+                return
+
+            # Get personas most relevant to the current session
+            top_personas = await chat_orchestrator.get_top_personas(
+                session_id=sid,
+            )
+
+            done_queue: asyncio.Queue = asyncio.Queue()
+
+            async def _run(pid: str) -> None:
+                try:
+                    persona = chat_orchestrator.get_persona(pid)
+                    result = await chat_orchestrator.generate_single_persona_response(
+                        session, persona,
+                        message.response_length or "medium",
+                    )
+                    session.append_message(pid, result["response"])
+                    await done_queue.put(result)
+                except Exception as e:
+                    logger.exception(f"chat-stream _run failed for {pid}: {e}")
+                    await done_queue.put({
+                        "persona_id": persona.id,
+                        "persona_name": persona.name,
+                        "response": f"I ran into a technical issue. Please try again. ({e!s})",
+                        "used_documents": False,
+                        "document_chunks_used": 0,
+                    })
+
+            tasks = [asyncio.create_task(_run(pid)) for pid in top_personas]
+
+            for _ in range(len(tasks)):
+                result = await done_queue.get()
+                line = ChatStreamLine(
+                    type="advisor",
+                    data={
+                        "persona_id": result["persona_id"],
+                        "persona_name": result["persona_name"],
+                        "content": result["response"],
+                        "used_documents": result.get("used_documents", False),
+                        "document_chunks_used": result.get("document_chunks_used", 0),
+                    },
+                )
+                yield line.to_ndjson()
+
+            await asyncio.gather(*tasks, return_exceptions=True)
+
+            yield ChatStreamLine(
+                type="progress",
+                data={"phase": "complete"},
+            ).to_ndjson()
+
+        except Exception as exc:
+            logger.error(f"chat-stream error: {exc}")
+            logger.error(traceback.format_exc())
+            yield ChatStreamLine(
+                type="error",
+                data={"detail": str(exc)},
+            ).to_ndjson()
+
+    return StreamingResponse(
+        _event_generator(),
+        media_type="application/x-ndjson",
+        headers={
+            "Cache-Control": "no-cache",
+            "X-Accel-Buffering": "no",
+        },
+    )
+
+
 @router.post("/switch-chat")
 async def switch_to_chat(
     request: SwitchChatRequest, 
@@ -157,9 +297,13 @@ async def chat_sequential_enhanced(
     message: ChatMessage, 
     request: Request,
     current_user: User = Depends(get_current_active_user)
-):
+    ) -> Dict[str, Any]:
     """
     Enhanced sequential chat with proper session management, document access, and intelligent persona ordering
+    @param message: ChatMessage containing user input and optional session/chat IDs
+    @param request: FastAPI Request object for session management
+    @param current_user: Authenticated user from dependency injection
+    @return: Dict response with LLM responses if successful, else error details
     """
     try:
         # Ensure consistent session ID for document retrieval
@@ -205,7 +349,7 @@ async def chat_sequential_enhanced(
         session.append_message("user", message.user_input)
         
         # Check if the user's message is vague and needs clarification
-        if chat_orchestrator._needs_clarification(session, message.user_input):
+        if chat_orchestrator.needs_clarification(session, message.user_input):
             clarification = await chat_orchestrator.generate_contextual_clarification(
                 message.user_input
             )

multi_llm_chatbot_backend/app/core/improved_orchestrator.py

@@ -49,7 +49,7 @@ class ImprovedChatOrchestrator:
             session.append_message("user", user_input)
             
             # Determine if we need clarification
-            needs_clarification = self._needs_clarification(session, user_input)
+            needs_clarification = self.needs_clarification(session, user_input)
             
             if needs_clarification:
                 # Generate clarification question
@@ -64,7 +64,7 @@ class ImprovedChatOrchestrator:
                 }
             
             # Generate responses from all personas
-            responses = await self._generate_persona_responses(session, response_length)
+            responses = await self.generate_persona_responses(session, response_length)
             
             return {
                 "status": "success",
@@ -100,7 +100,7 @@ class ImprovedChatOrchestrator:
             available_documents = [doc["filename"] for doc in doc_stats.get("documents", [])]
             
             # Generate enhanced persona responses
-            responses = await self._generate_persona_responses(session, response_length)
+            responses = await self.generate_persona_responses(session, response_length)
             
             return {
                 "status": "success",
@@ -136,7 +136,7 @@ class ImprovedChatOrchestrator:
         
         return references[:3]  # Limit to first 3 references
     
-    def _needs_clarification(self, session: ConversationContext, user_input: str) -> bool:
+    def needs_clarification(self, session: ConversationContext, user_input: str) -> bool:
         """
         Determine if the user input needs clarification.
         Patterns and keywords are driven by config.yaml → orchestrator section.
@@ -253,7 +253,7 @@ class ImprovedChatOrchestrator:
             "suggestions": fallback_suggestions,
         }
     
-    async def _generate_persona_responses(self, session: ConversationContext, response_length: str = "medium"):
+    async def generate_persona_responses(self, session: ConversationContext, response_length: str = "medium"):
         """
         Generate responses from all personas with enhanced RAG integration
         """
@@ -263,7 +263,7 @@ class ImprovedChatOrchestrator:
             logger.info(f"Generating response for {persona_id} with enhanced RAG")
             
             # Generate persona response with enhanced RAG
-            response_data = await self._generate_single_persona_response(session, persona, response_length)
+            response_data = await self.generate_single_persona_response(session, persona, response_length)
             
             # Add persona response to session context
             session.append_message(persona_id, response_data["response"])
@@ -272,7 +272,7 @@ class ImprovedChatOrchestrator:
         
         return responses
     
-    async def _generate_single_persona_response(self, session, persona, response_length: str = "medium"):
+    async def generate_single_persona_response(self, session, persona, response_length: str = "medium"):
         """
         Enhanced version - Generate response from a single persona with enhanced RAG integration
         """
@@ -700,7 +700,7 @@ When analyzing the document context:
             logger.info(f"Generating response for {persona_id} with session {session_id}")
             
             # Generate response from single persona using consistent session ID
-            response_data = await self._generate_single_persona_response(session, persona, response_length)
+            response_data = await self.generate_single_persona_response(session, persona, response_length)
             
             # Add response to session
             session.append_message(persona_id, response_data["response"])
@@ -771,6 +771,9 @@ When analyzing the document context:
                 for p in self.personas.values()
             ])
 
+            # Ensure k does not exceed the number of available personas
+            k = min(k, len(self.personas))
+
             app_title = get_settings().app.title
 
             prompt = f"""

phd-advisor-frontend/src/pages/ChatPage.js

@@ -396,17 +396,12 @@ const handleNewChat = async (sessionId = null) => {
     setThinkingAdvisors(['system']);
 
     try {
-      const headers = {
-        'Content-Type': 'application/json',
-        'Authorization': `Bearer ${authToken}`,
-      };
-      
-
-      console.log('Sending message with session ID:', currentSessionId); // Debug log
-
-      const response = await fetch(`${process.env.REACT_APP_API_URL}/chat-sequential`, {
+      const response = await fetch(`${process.env.REACT_APP_API_URL}/chat-stream`, {
         method: 'POST',
-        headers: headers,
+        headers: {
+          'Content-Type': 'application/json',
+          'Authorization': `Bearer ${authToken}`,
+        },
         body: JSON.stringify({
           user_input: inputMessage,
           response_length: 'medium',
@@ -418,32 +413,69 @@ const handleNewChat = async (sessionId = null) => {
         throw new Error(`HTTP error! status: ${response.status}`);
       }
 
-      // Rest of the function remains the same...
-      const data = await response.json();
-      
-      // Process responses...
-      if (data.responses && Array.isArray(data.responses)) {
-        const newResponses = data.responses.map(response => ({
-          id: generateMessageId(),
-          type: 'advisor',
-          persona_id: response.persona_id,
-          content: response.content,
-          timestamp: new Date(),
-          advisorName: response.persona_name || response.persona_id,
-          used_documents: response.used_documents || false,
-          document_chunks_used: response.document_chunks_used || 0
-        }));
-
-        setMessages(prev => [...prev, ...newResponses]);
-        
-        // Save advisor responses to database
-        for (const response of newResponses) {
-          await saveMessageToSession(response);
-        }
-        
-        // Log session debug info if available
-        if (data.session_debug) {
-          console.log('Session debug info:', data.session_debug);
+      const reader = response.body.getReader();
+      const decoder = new TextDecoder();
+      let buffer = '';
+
+      while (true) {
+        const { done, value } = await reader.read();
+        if (done) break;
+
+        buffer += decoder.decode(value, { stream: true });
+        const lines = buffer.split('\n');
+        buffer = lines.pop() ?? '';
+
+        for (const line of lines) {
+          if (!line.trim()) continue;
+          const payload = JSON.parse(line);
+
+          const d = payload.data || {};
+
+          switch (payload.type) {
+            case 'advisor': {
+              const msg = {
+                id: generateMessageId(),
+                type: 'advisor',
+                persona_id: d.persona_id,
+                content: d.content,
+                timestamp: new Date(),
+                advisorName: d.persona_name || d.persona_id,
+                used_documents: d.used_documents || false,
+                document_chunks_used: d.document_chunks_used || 0,
+              };
+              setMessages(prev => [...prev, msg]);
+              setThinkingAdvisors(prev => prev.filter(a => a !== d.persona_id));
+              await saveMessageToSession(msg);
+              break;
+            }
+            case 'clarification':
+              setMessages(prev => [...prev, {
+                id: generateMessageId(),
+                type: 'clarification',
+                content: d.message,
+                suggestions: d.suggestions || [],
+                timestamp: new Date(),
+              }]);
+              break;
+            case 'progress':
+              if (d.phase === 'complete') {
+                break;
+              }
+              if (d.persona_id != null) {
+                setThinkingAdvisors(prev => prev.filter(a => a !== d.persona_id));
+              }
+              break;
+            case 'error':
+              setMessages(prev => [...prev, {
+                id: generateMessageId(),
+                type: 'error',
+                content: d.detail || 'An error occurred',
+                timestamp: new Date(),
+              }]);
+              break;
+            default:
+              break;
+          }
         }
       }