Phase 3: streaming chat via SSE with prompt caching preserved

app.py

@@ -26,8 +26,10 @@ from claude_engine import claude_engine  # noqa: E402
 from opportunities_service import opportunities_service  # noqa: E402
 
 from fastapi import UploadFile, File, Form, HTTPException  # noqa: E402
+from fastapi.responses import StreamingResponse  # noqa: E402
 from pydantic import BaseModel  # noqa: E402
 from typing import Any, Dict, List, Optional  # noqa: E402
+import json as _json  # noqa: E402
 
 
 # ---------------------------------------------------------------------------
@@ -201,6 +203,90 @@ async def chat_claude(request: ClaudeChatRequest):
     }
 
 
+# ---------------------------------------------------------------------------
+# Phase 3: Streaming chat — Server-Sent Events
+# ---------------------------------------------------------------------------
+
+@app.post("/api/chat/claude/stream")
+async def chat_claude_stream(request: ClaudeChatRequest):
+    """
+    Streaming version of /api/chat/claude. Returns a Server-Sent Events
+    (SSE) response. Each event has shape:
+
+        data: {"type": "chunk", "text": "..."}\\n\\n
+        data: {"type": "sources", "sources": [...]}\\n\\n
+        data: {"type": "done"}\\n\\n
+
+    On Claude failure the stream emits a single "error" event and the
+    client should fall back to /api/chat/claude (non-streaming) which
+    in turn falls back to Groq.
+    """
+    if not claude_engine.enabled:
+        raise HTTPException(status_code=503, detail="Claude is not configured")
+    if retrieval_engine is None:
+        raise HTTPException(status_code=503, detail="Retrieval engine unavailable")
+
+    # Retrieval is synchronous and happens BEFORE we start streaming so
+    # we can include sources in the first event the client receives.
+    try:
+        context_docs = retrieval_engine.retrieve_context(
+            query=request.message, role="student"
+        )
+    except Exception as e:
+        print(f"[FAIL] Retrieval error: {e}")
+        context_docs = []
+
+    sources = []
+    for doc in (context_docs or [])[:3]:
+        meta = getattr(doc, "metadata", None) or {}
+        sources.append(
+            {
+                "title": meta.get("title", "ALU Knowledge Base"),
+                "source": meta.get("source", "uploaded"),
+                "doc_id": meta.get("doc_id"),
+            }
+        )
+
+    def sse_pack(event: Dict[str, Any]) -> str:
+        return f"data: {_json.dumps(event, ensure_ascii=False)}\n\n"
+
+    def event_generator():
+        # Send sources first so the UI can render a 'Reading from...' hint.
+        yield sse_pack({"type": "sources", "sources": sources})
+
+        emitted_any = False
+        try:
+            for chunk in claude_engine.stream_response(
+                query=request.message,
+                context_docs=context_docs,
+                history=request.history,
+            ):
+                emitted_any = True
+                yield sse_pack({"type": "chunk", "text": chunk})
+        except Exception as e:
+            print(f"[FAIL] Stream generator error: {e}")
+            yield sse_pack({"type": "error", "message": "Stream interrupted"})
+            return
+
+        if not emitted_any:
+            yield sse_pack(
+                {"type": "error", "message": "Claude returned no content"}
+            )
+            return
+
+        yield sse_pack({"type": "done"})
+
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "X-Accel-Buffering": "no",  # disable proxy buffering
+            "Connection": "keep-alive",
+        },
+    )
+
+
 # ---------------------------------------------------------------------------
 # Phase 2: Application essay coach
 # ---------------------------------------------------------------------------

claude_engine.py

@@ -7,7 +7,7 @@ documents uploaded via /api/documents/upload.
 """
 import os
 import logging
-from typing import List, Dict, Any, Optional
+from typing import List, Dict, Any, Optional, Iterator
 
 logger = logging.getLogger(__name__)
 
@@ -238,6 +238,83 @@ class ClaudeEngine:
             logger.error(f"[FAIL] Claude API error: {e}")
             return None
 
+    # ------------------------------------------------------------------
+    # Streaming chat — yields text chunks as Claude generates them
+    # ------------------------------------------------------------------
+
+    def stream_response(
+        self,
+        query: str,
+        context_docs: Optional[List[Any]] = None,
+        history: Optional[List[Dict[str, str]]] = None,
+        max_tokens: int = DEFAULT_MAX_TOKENS,
+        temperature: float = DEFAULT_TEMPERATURE,
+    ) -> Iterator[str]:
+        """
+        Same RAG behaviour as `generate_response`, but yields text chunks
+        as Claude generates them. Caller is responsible for forwarding chunks
+        to the client (e.g. via Server-Sent Events).
+
+        Yields ONLY the incremental text. Caller should accumulate to get
+        the full response if needed.
+
+        On failure yields nothing — caller should treat empty stream as
+        "fall back to non-streaming engine".
+        """
+        if not self.enabled or not self.client:
+            return
+
+        context_block = self._format_context(context_docs or [])
+
+        system_blocks = [
+            {
+                "type": "text",
+                "text": SYSTEM_PROMPT,
+                "cache_control": {"type": "ephemeral"},
+            },
+            {
+                "type": "text",
+                "text": (
+                    "Use the following context from the ALU knowledge base "
+                    "to answer the student's question. If the context does "
+                    "not contain the answer, say so honestly and point to "
+                    "the official source.\n\n"
+                    f"=== CONTEXT ===\n{context_block}\n=== END CONTEXT ==="
+                ),
+                "cache_control": {"type": "ephemeral"},
+            },
+        ]
+
+        messages = self._to_anthropic_messages(history or [], query)
+
+        try:
+            with self.client.messages.stream(
+                model=self.model,
+                max_tokens=max_tokens,
+                temperature=temperature,
+                system=system_blocks,
+                messages=messages,
+            ) as stream:
+                for text_chunk in stream.text_stream:
+                    if text_chunk:
+                        yield text_chunk
+
+                # Log usage from the final message for cache-hit auditing.
+                final = stream.get_final_message()
+                usage = getattr(final, "usage", None)
+                if usage:
+                    logger.info(
+                        "[OK] Claude stream complete | tokens in=%d out=%d "
+                        "cache_read=%d cache_write=%d",
+                        getattr(usage, "input_tokens", 0),
+                        getattr(usage, "output_tokens", 0),
+                        getattr(usage, "cache_read_input_tokens", 0) or 0,
+                        getattr(usage, "cache_creation_input_tokens", 0) or 0,
+                    )
+        except Exception as e:
+            logger.error(f"[FAIL] Claude streaming error: {e}")
+            return
+
     # ------------------------------------------------------------------
     # Application essay coach
     # ------------------------------------------------------------------