GenAI RAG and Bonus

infra/.env

@@ -25,3 +25,11 @@ GEMINI_API_KEY=""
 GROQ_API_KEY=""
 MISTRAL_API_KEY=""
 COHERE_API_KEY=""
+
+NOTE_SERVICE_URL=http://note-service-app:8005
+CALENDAR_SERVICE_URL=http://calendar-service-app:8004
+CHECKLIST_SERVICE_URL=http://checklist-service-app:8003
+
+WEAVIATE_HOST=weaviate
+WEAVIATE_HTTP_PORT=8080
+WEAVIATE_GRPC_PORT=50051

infra/docker-compose.yml

@@ -88,6 +88,7 @@ services:
       - "${CHECKLIST_SERVICE_APP_PORT}:${CHECKLIST_SERVICE_APP_PORT}"
     networks:
       - checklist-service-net
+      - internal-net
     depends_on:
       services-db:
         condition: service_healthy
@@ -105,6 +106,7 @@ services:
       - "${CALENDAR_SERVICE_APP_PORT}:${CALENDAR_SERVICE_APP_PORT}"
     networks:
       - calendar-service-net
+      - internal-net
     depends_on:
       services-db:
         condition: service_healthy
@@ -122,6 +124,7 @@ services:
       - "${NOTE_SERVICE_APP_PORT}:${NOTE_SERVICE_APP_PORT}"
     networks:
       - note-service-net
+      - internal-net
     depends_on:
       services-db:
         condition: service_healthy
@@ -148,6 +151,24 @@ services:
       services-db:
         condition: service_healthy
 
+  weaviate:
+    image: cr.weaviate.io/semitechnologies/weaviate:1.27.0
+    container_name: weaviate
+    ports:
+      - "8080:8080"
+      - "50051:50051"
+    environment:
+      QUERY_DEFAULTS_LIMIT: 25
+      AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED: "true"
+      PERSISTENCE_DATA_PATH: /var/lib/weaviate
+      DEFAULT_VECTORIZER_MODULE: none
+      ENABLE_MODULES: ""
+      CLUSTER_HOSTNAME: node1
+    volumes:
+      - weaviate-data:/var/lib/weaviate
+    networks:
+      - genai-service-net
+
   genai-service-app:
     <<: *app-template
     image: ghcr.io/aet-devops26/team-devopss26/genai-service:7435acd8700e13487e824899ffaa5df167e53792
@@ -158,9 +179,12 @@ services:
       - "${GENAI_SERVICE_APP_PORT}:${GENAI_SERVICE_APP_PORT}"
     networks:
       - genai-service-net
+      - internal-net
     depends_on:
       genai-service-liquibase:
         condition: service_completed_successfully
+      weaviate:
+        condition: service_started
     healthcheck:
       test: ["CMD-SHELL", "curl -f http://localhost:${GENAI_SERVICE_APP_PORT}/health || exit 1"]
       interval: 10s
@@ -184,6 +208,7 @@ services:
 # --- VOLUMES & NETWORKS ---
 # Define the volumes and networks used by the services.
 volumes:
+  weaviate-data: {}
   services-db: {}
 
 networks:
@@ -193,4 +218,5 @@ networks:
   calendar-service-net: {}
   note-service-net: {}
   genai-service-net: {}
-  client-net: {}
+  client-net: {}
+  internal-net: {}

services/calendar-service/src/main/java/de/tum/devopss26/calendarservice/CalendarEventController.java

@@ -18,8 +18,12 @@ public class CalendarEventController {
     @Operation(summary = "Get all events for a user")
     public ResponseEntity<List<CalendarEvent>> getAllEvents(@RequestParam Long userId) {
         return ResponseEntity.ok(List.of(
-                new CalendarEvent(1L, "Sample Event", "Sample description",
-                        LocalDateTime.now(), LocalDateTime.now().plusHours(1), "Munich")
+                new CalendarEvent(1L, "Car Service Appointment", "Oil change and brake inspection",
+                        LocalDateTime.of(2026, 6, 19, 10, 0), LocalDateTime.of(2026, 6, 19, 11, 0), "AutoShop Central"),
+                new CalendarEvent(2L, "Dentist Appointment", "Routine cleaning and checkup",
+                        LocalDateTime.of(2026, 6, 21, 14, 30), LocalDateTime.of(2026, 6, 21, 15, 30), "City Dental Clinic"),
+                new CalendarEvent(3L, "Grocery Run", "Weekly grocery shopping",
+                        LocalDateTime.of(2026, 6, 20, 9, 0), LocalDateTime.of(2026, 6, 20, 10, 0), "Supermarket")
         ));
     }
 

services/checklist-service/src/main/java/de/tum/devopss26/checklistservice/ChecklistController.java

@@ -16,7 +16,24 @@ public class ChecklistController {
     @GetMapping
     @Operation(summary = "Get all checklists for a user")
     public ResponseEntity<List<Checklist>> getAllChecklists(@RequestParam Long userId) {
-        return ResponseEntity.ok(List.of(new Checklist(1L, "Sample Checklist")));
+        Checklist groceries = new Checklist(1L, "Grocery Shopping");
+        groceries.setItems(List.of(
+                new ChecklistItem(1L, "Milk", true, 1),
+                new ChecklistItem(2L, "Eggs", false, 2),
+                new ChecklistItem(3L, "Bread", false, 3),
+                new ChecklistItem(4L, "Coffee", false, 4),
+                new ChecklistItem(5L, "Orange juice", false, 5)
+        ));
+
+        Checklist errands = new Checklist(2L, "Weekly Errands");
+        errands.setItems(List.of(
+                new ChecklistItem(6L, "Drop off dry cleaning", true, 1),
+                new ChecklistItem(7L, "Renew car insurance", false, 2),
+                new ChecklistItem(8L, "Pay electricity bill", false, 3),
+                new ChecklistItem(9L, "Return library books", false, 4)
+        ));
+
+        return ResponseEntity.ok(List.of(groceries, errands));
     }
 
     @GetMapping("/{id}")

services/genai-service/main.py

@@ -1,23 +1,30 @@
+import asyncio
 import os
 import time
+import uuid
+from contextlib import asynccontextmanager
 from datetime import datetime
 from typing import Optional, List, Literal
 
+import httpx
 import jwt
+import weaviate
 from fastapi import FastAPI, HTTPException, Depends
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
 from langchain_cohere import ChatCohere
 from langchain_core.output_parsers import StrOutputParser
 from langchain_core.prompts import ChatPromptTemplate
 from langchain_google_genai import ChatGoogleGenerativeAI
+from sentence_transformers import SentenceTransformer
 from langchain_groq import ChatGroq
 from langchain_mistralai import ChatMistralAI
 from pydantic import BaseModel
 from sqlalchemy import String, Text, ForeignKey, DateTime, BigInteger, CheckConstraint, func
 from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
 from sqlalchemy.future import select
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, selectinload
+from weaviate.classes.config import Property, DataType, Configure
 
 
 # ── Database ─────────────────────────────────────────────────────────────────
@@ -74,7 +81,7 @@ def _get_current_user_id(credentials: Optional[HTTPAuthorizationCredentials] = D
     try:
         if _JWT_PUBLIC_KEY:
             payload = jwt.decode(credentials.credentials, _JWT_PUBLIC_KEY, algorithms=["RS256"])
-        else:   # until security is fully implemented
+        else:
             payload = jwt.decode(credentials.credentials, options={"verify_signature": False}, algorithms=["RS256"])
     except jwt.ExpiredSignatureError:
         raise HTTPException(status_code=401, detail="Token has expired")
@@ -90,27 +97,131 @@ def _get_current_user_id(credentials: Optional[HTTPAuthorizationCredentials] = D
         raise HTTPException(status_code=401, detail="Token user identity is not a valid integer")
 
 
+# ── RAG: Weaviate + embeddings ────────────────────────────────────────────────
+_weaviate_client: Optional[weaviate.WeaviateClient] = None
+_embedding_model: Optional[SentenceTransformer] = None
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    global _weaviate_client, _embedding_model
+
+    try:
+        _weaviate_client = weaviate.connect_to_custom(
+            http_host=os.environ.get("WEAVIATE_HOST", "weaviate"),
+            http_port=int(os.environ.get("WEAVIATE_HTTP_PORT", "8080")),
+            http_secure=False,
+            grpc_host=os.environ.get("WEAVIATE_HOST", "weaviate"),
+            grpc_port=int(os.environ.get("WEAVIATE_GRPC_PORT", "50051")),
+            grpc_secure=False,
+        )
+    except Exception:
+        _weaviate_client = None
+
+    _embedding_model = SentenceTransformer("all-MiniLM-L6-v2")
+
+    yield
+
+    if _weaviate_client:
+        _weaviate_client.close()
+
+
+async def _fetch_user_data(user_id: int) -> list[str]:
+    note_url = os.environ.get("NOTE_SERVICE_URL", "http://note-service-app:8005")
+    calendar_url = os.environ.get("CALENDAR_SERVICE_URL", "http://calendar-service-app:8004")
+    checklist_url = os.environ.get("CHECKLIST_SERVICE_URL", "http://checklist-service-app:8003")
+
+    async with httpx.AsyncClient(timeout=10.0) as client:
+        notes_resp, events_resp, checklists_resp = await asyncio.gather(
+            client.get(f"{note_url}/api/v1/notes", params={"userId": user_id}),
+            client.get(f"{calendar_url}/api/v1/events", params={"userId": user_id}),
+            client.get(f"{checklist_url}/api/v1/checklists", params={"userId": user_id}),
+            return_exceptions=True,
+        )
+
+    chunks = []
+
+    if not isinstance(notes_resp, Exception) and notes_resp.status_code == 200:
+        for note in notes_resp.json():
+            title = note.get("title", "")
+            content = note.get("content", "")
+            if title or content:
+                chunks.append(f"[Note] {title}: {content}")
+
+    if not isinstance(events_resp, Exception) and events_resp.status_code == 200:
+        for event in events_resp.json():
+            chunks.append(
+                f"[Calendar Event] {event.get('title', '')} "
+                f"from {event.get('startTime', '')} to {event.get('endTime', '')} "
+                f"at {event.get('location', '')}: {event.get('description', '')}"
+            )
+
+    if not isinstance(checklists_resp, Exception) and checklists_resp.status_code == 200:
+        for checklist in checklists_resp.json():
+            title = checklist.get("title", "")
+            for item in checklist.get("items", []):
+                status = "completed" if item.get("completed") else "not completed"
+                chunks.append(f"[Checklist '{title}'] {item.get('text', '')} ({status})")
+
+    return chunks
+
+
+def _rag_sync(query: str, chunks: list[str], top_k: int) -> str:
+    all_vecs = _embedding_model.encode(chunks + [query])
+    chunk_vecs = all_vecs[:-1].tolist()
+    query_vec = all_vecs[-1].tolist()
+
+    collection_name = f"Session{uuid.uuid4().hex}"
+    try:
+        collection = _weaviate_client.collections.create(
+            name=collection_name,
+            properties=[Property(name="text", data_type=DataType.TEXT)],
+            vectorizer_config=Configure.Vectorizer.none(),
+        )
+        with collection.batch.dynamic() as batch:
+            for text, vector in zip(chunks, chunk_vecs):
+                batch.add_object(properties={"text": text}, vector=vector)
+        results = collection.query.near_vector(
+            near_vector=query_vec,
+            limit=min(top_k, len(chunks)),
+        )
+        return "\n".join(obj.properties["text"] for obj in results.objects)
+    finally:
+        _weaviate_client.collections.delete(collection_name)
+
+
+async def _rag_retrieve(query: str, chunks: list[str], top_k: int = 5) -> str:
+    if not chunks or _weaviate_client is None or _embedding_model is None:
+        return ""
+    loop = asyncio.get_running_loop()
+    return await loop.run_in_executor(None, _rag_sync, query, chunks, top_k)
+
+
 # ── LangChain ─────────────────────────────────────────────────────────────────
 _prompt = ChatPromptTemplate.from_messages([
-    ("system", "You are a helpful assistant for a personal productivity app. "
-               "Users can ask you about their notes, calendar events, and checklists."),
+    ("system",
+     "You are a helpful assistant for a personal productivity app. "
+     "Use the context below, retrieved from the user's personal notes, calendar events, and checklists, "
+     "to answer their question. If the context does not contain enough information, say so.\n\n"
+     "Context:\n{context}"),
     ("human", "{message}"),
 ])
 
+
 def _build_chain(model: str):
     if model == "groq-llama":
         llm = ChatGroq(model="llama-3.1-8b-instant", api_key=os.environ["GROQ_API_KEY"])
     elif model == "mistral":
         llm = ChatMistralAI(model="mistral-small-latest", api_key=os.environ["MISTRAL_API_KEY"])
     elif model == "cohere":
         llm = ChatCohere(model="command-r", cohere_api_key=os.environ["COHERE_API_KEY"])
-    else:  # default: gemini
+    else:
         llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", google_api_key=os.environ["GEMINI_API_KEY"])
     return _prompt | llm | StrOutputParser()
 
 
 # ── App ───────────────────────────────────────────────────────────────────────
-app = FastAPI(title="GenAI Chatbot Service", root_path="/api/v1")
+app = FastAPI(title="GenAI Chatbot Service", root_path="/api/v1", lifespan=lifespan)
 app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
 
 
@@ -180,7 +291,7 @@ async def create_conversation(
     )
     return result.scalar_one()
 
-# The schema of the output is defined above as ConversationOut
+
 @app.get("/conversations/{conversation_id}", response_model=ConversationOut)
 async def get_conversation(
     conversation_id: int,
@@ -206,22 +317,17 @@ async def delete_conversation(
     jwt_user_id: Optional[int] = Depends(_get_current_user_id),
     db: AsyncSession = Depends(get_db),
 ):
-
-    # Check whether the conversation actually exists
     result = await db.execute(select(ChatConversation).where(ChatConversation.id == conversation_id))
     conversation = result.scalar_one_or_none()
     if not conversation:
         raise HTTPException(status_code=404, detail="Conversation not found")
     if jwt_user_id is not None and conversation.user_id != jwt_user_id:
         raise HTTPException(status_code=403, detail="Access denied")
-
-    # Delete the conversation
     await db.delete(conversation)
     await db.commit()
     return {"message": "Conversation deleted"}
 
 
-## This is the main endpoint that chat requests go through
 @app.post("/chat", response_model=ChatResponse)
 async def chat(
     request: ChatRequest,
@@ -233,8 +339,7 @@ async def chat(
 
     user_id = jwt_user_id if jwt_user_id is not None else request.user_id
 
-    if request.conversation_id:             # If this is not a new conversation
-        # Check whether the conversation actually exists in the database
+    if request.conversation_id:
         result = await db.execute(
             select(ChatConversation).where(ChatConversation.id == request.conversation_id)
         )
@@ -244,19 +349,27 @@ async def chat(
         if jwt_user_id is not None and conversation.user_id != user_id:
             raise HTTPException(status_code=403, detail="Access denied")
     else:
-        # Otherwise create a new conversation
         conversation = ChatConversation(user_id=user_id, title=request.message[:100])
         db.add(conversation)
         await db.flush()
 
-    # Add the user's message to the conversation with the role "USER"
-    db.add(ChatMessage( 
+    db.add(ChatMessage(
         conversation_id=conversation.id, role="USER",
         content=request.message, timestamp=_now_ms(),
     ))
-    response_text = await _build_chain(request.model).ainvoke({"message": request.message})
 
-    # Add the model's message to the conversation with the role "AGENT"
+    context = ""
+    try:
+        chunks = await _fetch_user_data(user_id)
+        context = await _rag_retrieve(request.message, chunks)
+    except Exception:
+        pass
+
+    response_text = await _build_chain(request.model).ainvoke({
+        "message": request.message,
+        "context": context or "No relevant data found in the user's notes, calendar, or checklists.",
+    })
+
     db.add(ChatMessage(
         conversation_id=conversation.id, role="AGENT",
         content=response_text, timestamp=_now_ms(),

services/genai-service/requirements.txt

@@ -9,3 +9,6 @@ pydantic==2.9.0
 sqlalchemy[asyncio]==2.0.36
 asyncpg==0.29.0
 PyJWT[crypto]==2.9.0
+weaviate-client>=4.6.0
+httpx>=0.27.0
+sentence-transformers>=3.0.0