Deploy backend with correct structure

Dockerfile

@@ -0,0 +1,45 @@
+# === STAGE 1: The Builder (Downloads the model) ===
+FROM python:3.10-slim as builder
+RUN pip install --no-cache-dir sentence-transformers
+ENV MODEL_NAME=sentence-transformers/all-MiniLM-L6-v2
+ENV SAVE_PATH=/opt/models/all-MiniLM-L6-v2
+RUN python -c "import os; from sentence_transformers import SentenceTransformer; SentenceTransformer(os.getenv('MODEL_NAME')).save(os.getenv('SAVE_PATH'))"
+
+# === STAGE 2: The Final Application Image ===
+FROM python:3.11-slim
+
+# Set environment variables for writable cache directories (matching your reference)
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    NLTK_DATA=/tmp/nltk_data \
+    TRANSFORMERS_CACHE=/tmp/transformers_cache \
+    HF_HOME=/tmp/huggingface \
+    PYTHONPATH=/app
+
+# Set work directory
+WORKDIR /app
+
+# Install system dependencies (if needed)
+RUN apt-get update && apt-get install -y \
+    build-essential \
+    && rm -rf /var/lib/apt/lists/*
+
+# Create writable cache directories
+RUN mkdir -p /tmp/nltk_data /tmp/transformers_cache /tmp/huggingface
+
+# Install Python dependencies
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy your application code (exactly like your reference)
+COPY main.py .
+COPY api/ ./api/
+
+# Copy the pre-downloaded model from the builder stage
+COPY --from=builder /opt/models/all-MiniLM-L6-v2 ./models/all-MiniLM-L6-v2
+
+# Expose the port
+EXPOSE 7860
+
+# Run the application (exactly like your reference)
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -0,0 +1,4 @@
+---
+title: Research Flowstream (Backend)
+sdk: docker
+---

api/crew.py

@@ -0,0 +1,184 @@
+import os
+import uuid
+import asyncio
+import json
+import requests
+from typing import AsyncGenerator
+from dotenv import load_dotenv
+
+# Load .env so env vars are available when starting Uvicorn directly
+load_dotenv()
+
+# Groq configuration
+GROQ_API_KEY = os.getenv("GROQ_API_KEY", "").strip()
+GROQ_MODEL = os.getenv("GROQ_MODEL", "llama-3.1-8b-instant").strip()
+GROQ_URL = "https://api.groq.com/openai/v1/chat/completions"
+
+# Optional switch to simulate local behavior (no external calls)
+GROQ_DISABLED = os.getenv("GROQ_DISABLED", "").lower() in {"1", "true", "yes"}
+
+# Reasonable connect/read timeouts for generation/streaming
+DEFAULT_TIMEOUT = (10, 120)
+
+# Base headers for Groq API
+HEADERS = {
+    "Authorization": f"Bearer {GROQ_API_KEY}" if GROQ_API_KEY else "",
+    "Content-Type": "application/json",
+}
+
+
+def generate_report_id() -> str:
+    """Create a unique ID for each report."""
+    return str(uuid.uuid4())
+
+
+def stream_event(kind: str, data):
+    """
+    Serialize events as proper JSON for SSE.
+    The FastAPI route will send lines like: `data: <json>\n\n`
+    Frontend can safely parse with json.loads(payload).
+    """
+    return json.dumps({"kind": kind, "data": data}, ensure_ascii=False)
+
+
+def _chunk(text: str, n: int):
+    """Split text into small pieces to render a smoother streaming experience."""
+    for i in range(0, len(text), n):
+        yield text[i : i + n]
+
+
+async def run_researcher_async(topic: str) -> str:
+    """
+    Researcher step: produce compact factual bullets.
+    Fallback text is returned if GROQ is disabled or unavailable.
+    """
+    if GROQ_DISABLED or not GROQ_API_KEY:
+        return (
+            f"- What is '{topic}'?\n"
+            f"- 3–5 key facts\n"
+            f"- Common use cases\n"
+            f"- Simple examples\n"
+        )
+
+    payload = {
+        "model": GROQ_MODEL,
+        "messages": [
+            {"role": "system", "content": "You are a concise researcher."},
+            {
+                "role": "user",
+                "content": f"Provide compact, factual bullet points about '{topic}'. "
+                           f"Max 8 bullets. Avoid filler text.",
+            },
+        ],
+        "temperature": 0.5,
+    }
+    try:
+        r = requests.post(GROQ_URL, headers=HEADERS, json=payload, timeout=DEFAULT_TIMEOUT)
+        r.raise_for_status()
+        return r.json()["choices"][0]["message"]["content"]
+    except Exception as e:
+        # Fallback on any network/API error
+        return f"[fallback researcher due to error: {e}]\n- Background\n- Key points\n- Examples"
+
+
+async def run_analyst_async(researcher_notes: str) -> str:
+    """
+    Analyst step: extract key insights and implications from researcher notes.
+    Fallback text is returned if GROQ is disabled or unavailable.
+    """
+    if GROQ_DISABLED or not GROQ_API_KEY:
+        return "- 3 key insights\n- 2 implications\n- 1 trade-off\n"
+
+    payload = {
+        "model": GROQ_MODEL,
+        "messages": [
+            {"role": "system", "content": "You extract insights cleanly."},
+            {
+                "role": "user",
+                "content": f"From these notes, produce exactly 3 insights and 2 implications:\n{researcher_notes}",
+            },
+        ],
+        "temperature": 0.5,
+    }
+    try:
+        r = requests.post(GROQ_URL, headers=HEADERS, json=payload, timeout=DEFAULT_TIMEOUT)
+        r.raise_for_status()
+        return r.json()["choices"][0]["message"]["content"]
+    except Exception as e:
+        return f"[fallback analyst due to error: {e}]\n- Insight 1\n- Insight 2\n- Insight 3\n- Implication A\n- Implication B"
+
+
+async def run_writer_token_stream(
+    topic: str,
+    researcher_notes: str,
+    analyst_notes: str,
+) -> AsyncGenerator[str, None]:
+    """
+    Writer step: stream the final report as small token-like chunks for smooth UI updates.
+    Yields strings (small chunks). Caller accumulates or forwards as SSE tokens.
+    """
+    writer_prompt = (
+        "Write a clear, beginner-friendly report with markdown headings:\n"
+        "Sections: Introduction, Key Concepts, Insights, Practical Tips, Conclusion.\n"
+        "Use concise language and bullets where helpful.\n\n"
+        f"Topic: {topic}\n\n"
+        f"Researcher Notes:\n{researcher_notes}\n\n"
+        f"Analyst Notes:\n{analyst_notes}\n"
+    )
+
+    # Local simulated streaming if GROQ is disabled or key missing
+    if GROQ_DISABLED or not GROQ_API_KEY:
+        simulated = [
+            f"## {topic}\n\n",
+            "### Introduction\n",
+            "This response is streaming locally to simulate real-time typing.\n\n",
+            "### Key Concepts\n",
+            "- Concept A\n- Concept B\n\n",
+            "### Insights\n",
+            "- Insight 1\n- Insight 2\n\n",
+            "### Practical Tips\n",
+            "- Tip 1\n- Tip 2\n\n",
+            "### Conclusion\n",
+            "Short summary.\n",
+        ]
+        for piece in simulated:
+            for small in _chunk(piece, 20):
+                yield small
+                await asyncio.sleep(0.015)
+        return
+
+    # Real streaming via Groq's OpenAI-compatible API
+    payload = {
+        "model": GROQ_MODEL,
+        "messages": [
+            {"role": "system", "content": "You are a clear, helpful technical writer."},
+            {"role": "user", "content": writer_prompt},
+        ],
+        "temperature": 0.6,
+        "stream": True,
+    }
+
+    # Using requests stream; iterate server-sent "data: ..." lines
+    with requests.post(
+        GROQ_URL, headers=HEADERS, json=payload, stream=True, timeout=DEFAULT_TIMEOUT
+    ) as resp:
+        resp.raise_for_status()
+        for line in resp.iter_lines(decode_unicode=True):
+            if not line:
+                continue
+            if not line.startswith("data: "):
+                continue
+            data = line[6:].strip()
+            if data == "[DONE]":
+                break
+            try:
+                obj = json.loads(data)
+                delta = obj["choices"][0]["delta"].get("content", "")
+                if not delta:
+                    continue
+                # Yield tiny chunks to update UI frequently
+                for small in _chunk(delta, 20):
+                    yield small
+            except Exception:
+                # Skip malformed lines gracefully
+                continue

api/db.py

@@ -0,0 +1,73 @@
+import os
+from typing import List, Dict, Any
+from dotenv import load_dotenv
+from qdrant_client import QdrantClient
+from qdrant_client.models import PointStruct, VectorParams, Distance
+from sentence_transformers import SentenceTransformer
+
+# Load environment variables from a .env file if it exists
+load_dotenv()
+
+# --- Qdrant Configuration ---
+QDRANT_URL = os.getenv("QDRANT_URL", "http://localhost:6333").strip()
+QDRANT_API_KEY = os.getenv("QDRANT_API_KEY", "").strip()
+COLLECTION = "reports"
+
+# --- Model Loading (from local files within the Docker image) ---
+# This path corresponds to where the Dockerfile copies the model.
+MODEL_PATH = "./models/all-MiniLM-L6-v2" 
+
+# Initialize the embedding model from the local path.
+try:
+    print(f"Loading sentence-transformer model from local path: {MODEL_PATH}")
+    embedding_model = SentenceTransformer(MODEL_PATH)
+    print("✅ Model loaded successfully!")
+except Exception as e:
+    print(f"❌ FATAL: Could not load the embedding model from {MODEL_PATH}.")
+    print("This indicates an issue with the Docker build or the file path in db.py.")
+    raise e
+
+# --- Qdrant Client and Collection Setup ---
+def _make_client() -> QdrantClient:
+    """Creates a Qdrant client based on environment variables."""
+    if QDRANT_API_KEY:
+        return QdrantClient(url=QDRANT_URL, api_key=QDRANT_API_KEY, timeout=30.0, check_compatibility=False)
+    else:
+        return QdrantClient(url=QDRANT_URL, timeout=30.0, check_compatibility=False)
+
+qdrant = _make_client()
+
+def _ensure_collection():
+    """Ensures the Qdrant collection exists, creating it if necessary."""
+    try:
+        qdrant.get_collection(collection_name=COLLECTION)
+        print(f"✅ Collection '{COLLECTION}' already exists.")
+    except Exception:
+        print(f"🔧 Collection '{COLLECTION}' not found. Creating it...")
+        qdrant.create_collection(
+            collection_name=COLLECTION,
+            vectors_config=VectorParams(size=384, distance=Distance.COSINE),
+        )
+        print("✅ Collection created.")
+
+_ensure_collection()
+
+# --- Database Functions ---
+def save_report(report_id: str, text: str, title: str):
+    """Encodes and saves a report to Qdrant."""
+    vector = embedding_model.encode(text).tolist()
+    qdrant.upsert(
+        collection_name=COLLECTION,
+        points=[PointStruct(id=report_id, vector=vector, payload={"text": text, "title": title})],
+    )
+
+def list_reports() -> List[Dict[str, Any]]:
+    """Lists recent reports, including their titles."""
+    hits, _ = qdrant.scroll(collection_name=COLLECTION, limit=50)
+    return [{"id": h.id, "title": h.payload.get("title", "(untitled)"), "text": h.payload.get("text", "")} for h in hits]
+
+def search_reports(query: str) -> List[Dict[str, Any]]:
+    """Performs semantic search and returns reports with titles."""
+    vector = embedding_model.encode(query).tolist()
+    hits = qdrant.search(collection_name=COLLECTION, query_vector=vector, limit=5)
+    return [{"id": hit.id, "score": float(hit.score), "title": hit.payload.get("title", "(untitled)"), "text": hit.payload.get("text", "")} for hit in hits]

api/models.py

@@ -0,0 +1,9 @@
+# Pydantic schemas
+from pydantic import BaseModel
+
+class ResearchRequest(BaseModel):
+    topic: str
+
+class SearchRequest(BaseModel):
+    query: str
+

api/routes.py

@@ -0,0 +1,53 @@
+from fastapi import APIRouter
+from fastapi.responses import StreamingResponse
+from .models import ResearchRequest, SearchRequest  # relative import
+from . import db, crew  # relative import
+
+router = APIRouter()
+
+@router.post("/start-job-stream")
+async def start_research_job_stream(request: ResearchRequest):
+    async def event_generator():
+        # Researcher
+        yield f"data: {crew.stream_event('stage', 'researcher:start')}\n\n"
+        researcher_notes = await crew.run_researcher_async(request.topic)
+        yield f"data: {crew.stream_event('stage', 'researcher:done')}\n\n"
+
+        # Analyst
+        yield f"data: {crew.stream_event('stage', 'analyst:start')}\n\n"
+        analyst_notes = await crew.run_analyst_async(researcher_notes)
+        yield f"data: {crew.stream_event('stage', 'analyst:done')}\n\n"
+
+        # Writer: token stream
+        yield f"data: {crew.stream_event('stage', 'writer:start')}\n\n"
+        final_accum = []
+        async for token in crew.run_writer_token_stream(
+            topic=request.topic,
+            researcher_notes=researcher_notes,
+            analyst_notes=analyst_notes,
+        ):
+            final_accum.append(token)
+            yield f"data: {crew.stream_event('token', token)}\n\n"
+
+        full_text = "".join(final_accum)
+        yield f"data: {crew.stream_event('stage', 'writer:done')}\n\n"
+
+        # Save exactly what was streamed
+        report_id = crew.generate_report_id()
+        db.save_report(report_id, full_text, title=request.topic)
+
+        # Final event (build dict first to avoid f-string brace issues)
+        meta = {"report_id": report_id, "title": request.topic}
+        yield f"data: {crew.stream_event('final', meta)}\n\n"
+        yield "event: close\ndata: done\n\n"
+
+    # Important for SSE: ensure streaming MIME and no buffering on your proxy
+    return StreamingResponse(event_generator(), media_type="text/event-stream")
+
+@router.get("/list-reports")
+async def list_reports():
+    return db.list_reports()
+
+@router.post("/search-reports")
+async def search_reports(request: SearchRequest):
+    return db.search_reports(request.query)

main.py

@@ -0,0 +1,23 @@
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+# Updated import - since 'api' folder will be directly under /app in the container
+from api.routes import router
+
+app = FastAPI(title="Multi-Agent Research Assistant Backend")
+
+# Register routes
+app.include_router(router)
+
+# CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # tighten in prod
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+@app.get("/")
+async def root():
+    return {"message": "Backend is running. Check /docs for API details."}

requirements.txt

@@ -0,0 +1,8 @@
+# Backend dependencies
+fastapi
+uvicorn
+qdrant-client
+sentence-transformers
+requests
+python-dotenv
+