Merge pull request #64 from AET-DevOps25/37-genai-rag-learning-path

37 genai rag learning path 
Closes #37

Author: mahdibayouli

genai/src/main.py

@@ -19,23 +19,25 @@
 from .services.embedding.schemas import EmbedRequest, EmbedResponse, QueryRequest, QueryResponse, DocumentResult
 from .services.embedding.weaviate_service import get_weaviate_client, ensure_schema_exists, DOCUMENT_CLASS_NAME
 from .services.llm import llm_service
-from .services.llm.schemas import GenerateRequest, GenerateResponse
-from .utils.error_schema import ErrorResponse
+from .services.llm.schemas import GenerateRequest, GenerateResponse 
+from .services.rag.schemas import CourseGenerationRequest, Course 
+from .services.rag import course_generator 
+from .utils.error_schema import ErrorResponse 
 from .utils.handle_httpx_exception import handle_httpx_exception
 
-
 # --- Configuration ---
 load_dotenv()
 logger = logging.getLogger("skillforge.genai")
 
 APP_PORT = int(os.getenv("GENAI_PORT", "8082"))
 APP_TITLE = os.getenv("GENAI_APP_NAME", "SkillForge GenAI Service")
 APP_VERSION = os.getenv("GENAI_APP_VERSION", "0.0.1")
-APP_DESCRIPTION = (
-    "SkillForge GenAI Service provides endpoints for web crawling, "
-    "chunking, embedding, semantic querying, and text generation using LLMs. "
-    "Ideal for integrating vector search and AI-driven workflows."
-)
+APP_DESCRIPTION = ( 
+    "SkillForge GenAI Service provides endpoints for web crawling, " 
+    "chunking, embedding, semantic querying, and text generation using LLMs. " 
+    "Ideal for integrating vector search and AI-driven workflows." 
+) 
+API_PREFIX = "/api/v1" 
 TAGS_METADATA = [
     {"name": "System", "description": "Health checks and system status."},
     {"name": "Crawler", "description": "Crawl and clean website content."},
@@ -110,7 +112,7 @@ async def unhandled_exception_handler(request: Request, exc: Exception):
 
 # ---- System Endpoints --------
 # -------------------------------
-@app.get("/health", tags=["System"])
+@app.get(f"{API_PREFIX}/health", tags=["System"])
 async def health():
     """
     Deep health check. Verifies the application and its core dependencies (e.g., DB, vector store).
@@ -126,7 +128,7 @@ async def health():
             content={"status": "error", "message": "Dependency failure. See logs for details."}
         )
 
-@app.get("/ping", tags=["System"])
+@app.get(f"{API_PREFIX}/ping", tags=["System"])
 async def ping():
     """
     Lightweight liveness check. Confirms the API process is running, but does not check dependencies.
@@ -139,7 +141,7 @@ async def ping():
 # -------------------------------
 # ----- Crawler endpoints -----
 # -------------------------------
-@app.post("/crawl", response_model=CrawlResponse, responses={400: {"model": ErrorResponse}, 500: {"model": ErrorResponse}}, tags=["Crawler"])
+@app.post(f"{API_PREFIX}/crawl", response_model=CrawlResponse, responses={400: {"model": ErrorResponse}, 500: {"model": ErrorResponse}}, tags=["Crawler"])
 async def crawl(request: CrawlRequest):
     url = str(request.url)
     try:
@@ -174,7 +176,7 @@ async def crawl(request: CrawlRequest):
 # -------------------------------
 # ----- Vector DB endpoints -----
 # -------------------------------
-@app.post("/embed", response_model=EmbedResponse, tags=["Embedder"])
+@app.post(f"{API_PREFIX}/embed", response_model=EmbedResponse, tags=["Embedder"])
 async def embed_url(request: EmbedRequest):
     """Orchestrates the full workflow: Crawl -> Chunk -> Embed -> Store."""
     url_str = str(request.url)
@@ -209,7 +211,7 @@ async def embed_url(request: EmbedRequest):
 
 
 
-@app.post("/query", response_model=QueryResponse)
+@app.post(f"{API_PREFIX}/query", response_model=QueryResponse)
 async def query_vector_db(request: QueryRequest):
     """Queries the vector database for text chunks semantically similar to the query."""
     client = get_weaviate_client()
@@ -231,7 +233,7 @@ async def query_vector_db(request: QueryRequest):
 # -------------------------------
 # --- LLM Endpoints -------------
 # -------------------------------
-@app.post("/generate", response_model=GenerateResponse, tags=["LLM"])
+@app.post(f"{API_PREFIX}/generate", response_model=GenerateResponse, tags=["LLM"])
 async def generate_completion(request: GenerateRequest):
     """Generates a text completion using the configured LLM abstraction layer."""
     try:
@@ -245,7 +247,19 @@ async def generate_completion(request: GenerateRequest):
         logging.error(f"ERROR during text generation: {e}")
         raise HTTPException(status_code=500, detail=f"Failed to generate text: {str(e)}")
 
-
+# ────────────────────────────────────────────────────────────────────────── 
+# NEW – main RAG endpoint 
+# ────────────────────────────────────────────────────────────────────────── 
+@app.post(f"{API_PREFIX}/rag/generate-course", response_model=Course, tags=["rag"]) 
+async def generate_course(req: CourseGenerationRequest): 
+    """ 
+    • POST because generation is a side-effectful operation (non-idempotent). 
+    • Returns a fully-validated Course JSON ready for the course-service. 
+    """ 
+    try: 
+        return course_generator.generate_course(req) 
+    except Exception as e: 
+        raise HTTPException(500, str(e)) from e
 
 # -------------------------------
 # --------- MAIN ----------------

genai/src/services/embedding/embedder_service.py

@@ -4,6 +4,9 @@
 from langchain_community.vectorstores.weaviate import Weaviate
 from .weaviate_service import get_weaviate_client, DOCUMENT_CLASS_NAME
 import logging
+from typing import List
+import numpy as np
+from .schemas import QueryResponse, QueryRequest, DocumentResult
 
 logger = logging.getLogger("skillforge.genai.embedder_service")
 
@@ -42,4 +45,34 @@ def embed_and_store_text(text: str, source_url: str) -> int:
     else:
         logger.info(f"Stored {num_chunks} chunks in Weaviate for URL {source_url}.")
 
-    return num_chunks
+    return num_chunks
+
+_embeddings_model = OpenAIEmbeddings(model="text-embedding-3-small")
+
+def embed_text(text: str) -> List[float]:
+    """Generate a single embedding vector from raw text."""
+    return _embeddings_model.embed_query(text)
+
+def cosine_similarity(v1: List[float], v2: List[float]) -> float:
+    """Simple cosine similarity between two vectors."""
+    a = np.array(v1)
+    b = np.array(v2)
+    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
+
+def query_similar_chunks(query_text: str, limit: int = 3) -> QueryResponse:
+    """
+    Stateless helper – identical logic to the /query endpoint but callable in-process.
+    """
+    client = get_weaviate_client()
+    embeddings_model = OpenAIEmbeddings(model="text-embedding-3-small")
+    vector = embeddings_model.embed_query(query_text)
+
+    result = (
+        client.query
+        .get(DOCUMENT_CLASS_NAME, ["content", "source_url"])
+        .with_near_vector({"vector": vector})
+        .with_limit(limit)
+        .do()
+    )
+    docs = [DocumentResult(**d) for d in result["data"]["Get"][DOCUMENT_CLASS_NAME]]
+    return QueryResponse(query=query_text, results=docs)

genai/src/services/embedding/schemas.py

@@ -1,4 +1,3 @@
-# genai/src/services/embedding/schemas.py
 from pydantic import BaseModel, HttpUrl
 from typing import List, Optional
 

genai/src/services/embedding/weaviate_service.py

@@ -6,7 +6,7 @@
 DOCUMENT_CLASS_NAME = "DocumentChunk"
 
 WEAVIATE_HOST = os.getenv("WEAVIATE_HOST", "localhost")
-WEAVIATE_HTTP_PORT = int(os.getenv("WEAVIATE_HTTP_PORT", "1234"))
+WEAVIATE_HTTP_PORT = int(os.getenv("WEAVIATE_HTTP_PORT", "8080"))
 WEAVIATE_GRPC_PORT = int(os.getenv("WEAVIATE_GRPC_PORT", "50051"))
 
 def get_weaviate_client() -> weaviate.Client:

genai/src/services/llm/llm_service.py

@@ -1,8 +1,15 @@
 import os
-import logging
 from langchain_openai import ChatOpenAI
+import json
+import logging
 from langchain_community.llms import FakeListLLM
 from langchain_core.language_models.base import BaseLanguageModel
+from typing import List, Type, TypeVar
+from pydantic import BaseModel, ValidationError
+
+logger = logging.getLogger(__name__)
+T = TypeVar("T", bound=BaseModel)
+
 
 def llm_factory() -> BaseLanguageModel:
     """
@@ -11,7 +18,7 @@ def llm_factory() -> BaseLanguageModel:
     Supports OpenAI, OpenAI-compatible (local/llmstudio), and dummy models.
     """
     provider = os.getenv("LLM_PROVIDER", "dummy").lower()
-    logging.info(f"--- Creating LLM for provider: {provider} ---")
+    logger.info(f"--- Creating LLM for provider: {provider} ---")
 
     if provider in ("openai", "llmstudio", "local"):
         # Get API base and key from env
@@ -59,4 +66,61 @@ def generate_text(prompt: str) -> str:
     if hasattr(response, 'content'):
         return response.content
     else:
-        return response
+        return response
+
+    
+def generate_structured(
+    messages: List[dict],
+    schema: Type[T],
+    *,
+    max_retries: int = 3,
+) -> T:
+    """Return a Pydantic object regardless of provider (OpenAI JSON-mode or fallback)."""
+    provider = os.getenv("LLM_PROVIDER", "dummy").lower()
+
+    # 1) OpenAI native JSON mode
+    if provider == "openai":
+        try:
+            from openai import OpenAI
+            client = OpenAI(
+                api_key=os.getenv("OPENAI_API_KEY"),
+                base_url=os.getenv("OPENAI_API_BASE", "https://api.openai.com/v1"),
+            )
+            resp = client.beta.chat.completions.parse(
+                model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"),
+                messages=messages,
+                response_format=schema,
+            )
+            return resp.choices[0].message.parsed  # type: ignore[arg-type]
+        except Exception as e:
+            logger.warning(f"OpenAI structured parse failed – falling back: {e}")
+
+    # 2) Generic JSON-string fallback
+    system_json_guard = {
+        "role": "system",
+        "content": (
+            "Return ONLY valid JSON matching this schema:\n"
+            + json.dumps(schema.model_json_schema())
+        ),
+    }
+    convo: List[dict] = [system_json_guard] + messages
+    llm = LLM_SINGLETON
+
+    for attempt in range(1, max_retries + 1):
+        raw = llm.invoke(convo)
+        text = raw.content if hasattr(raw, "content") else raw
+        try:
+            return schema.model_validate_json(text)
+        except ValidationError as e:
+            logger.warning(
+                f"Structured output validation failed ({attempt}/{max_retries}): {e}"
+            )
+            convo += [
+                {"role": "assistant", "content": text},
+                {
+                    "role": "user",
+                    "content": "❌ JSON invalid. Send ONLY fixed JSON.",
+                },
+            ]
+
+    raise ValueError("Could not obtain valid structured output")