Repository: kaito-project/kaito

Issue:
Add support for chat/completions API in RAGEngine
**Is your feature request related to a problem? Please describe.**
The API model of the RAGEngine should more closely resemble the [chat/completion](https://platform.openai.com/docs/api-reference/chat) spec.

**Describe the solution you'd like**
Update the RAGEngine `/query` API to support chat message formats. 

**Describe alternatives you've considered**

**Additional context**

Discussion:

Solution patch:
diff --git a/presets/ragengine/config.py b/presets/ragengine/config.py
index 387ef02..ae38585 100644
--- a/presets/ragengine/config.py
+++ b/presets/ragengine/config.py
@@ -51,6 +51,7 @@ LLM_RERANKER_TOP_N = int(os.getenv("LLM_RERANKER_TOP_N", 3))  # Default top 3 re
 # LLM (Large Language Model) configuration
 LLM_INFERENCE_URL = os.getenv("LLM_INFERENCE_URL", "http://localhost:5000/v1/completions")
 LLM_ACCESS_SECRET = os.getenv("LLM_ACCESS_SECRET", "default-access-secret")
+LLM_CONTEXT_WINDOW = int(os.getenv("LLM_CONTEXT_WINDOW", 64000))  # Default context window size
 # LLM_RESPONSE_FIELD = os.getenv("LLM_RESPONSE_FIELD", "result")  # Uncomment if needed in the future
 
 """
diff --git a/presets/ragengine/inference/inference.py b/presets/ragengine/inference/inference.py
index 2d57b44..f9f3629 100644
--- a/presets/ragengine/inference/inference.py
+++ b/presets/ragengine/inference/inference.py
@@ -16,16 +16,21 @@ from pydantic import PrivateAttr
 import logging
 import asyncio
 import httpx
-from typing import Any
-from llama_index.core.llms import CustomLLM, CompletionResponse, LLMMetadata, CompletionResponseGen
-from llama_index.llms.openai import OpenAI
-from llama_index.core.llms.callbacks import llm_completion_callback
+from typing import Any, Sequence
+from llama_index.core.llms import CustomLLM, CompletionResponse, LLMMetadata, CompletionResponseGen, ChatMessage, ChatResponse
+from llama_index.core.llms.callbacks import llm_completion_callback, llm_chat_callback
 import requests
 from requests.exceptions import HTTPError
 from urllib.parse import urlparse, urljoin
-from ragengine.config import LLM_INFERENCE_URL, LLM_ACCESS_SECRET #, LLM_RESPONSE_FIELD
+from ragengine.config import LLM_INFERENCE_URL, LLM_ACCESS_SECRET, LLM_CONTEXT_WINDOW #, LLM_RESPONSE_FIELD
+from ragengine.models import ChatCompletionResponse
 from fastapi import HTTPException
 import concurrent.futures
+import json
+
+from openai.types.chat import (
+    CompletionCreateParams,
+)
 
 # Configure logging
 logging.basicConfig(level=logging.INFO)
@@ -88,12 +93,7 @@ class Inference(CustomLLM):
     @llm_completion_callback()
     async def acomplete(self, prompt: str, formatted: bool = False, **kwargs: Any) -> CompletionResponse:
         try:
-            if LLM_INFERENCE_URL.startswith(OPENAI_URL_PREFIX):
-                return await self._async_openai_complete(prompt, **kwargs, **self.params)
-            elif LLM_INFERENCE_URL.startswith(HUGGINGFACE_URL_PREFIX):
-                return await self._async_huggingface_remote_complete(prompt, **kwargs, **self.params)
-            else:
-                return await self._async_custom_api_complete(prompt, **kwargs, **self.params)
+            return await self._async_completions(prompt, **kwargs, **self.params)
         except HTTPException as http_exc:
             raise http_exc
         except Exception as e:
@@ -102,14 +102,91 @@ class Inference(CustomLLM):
         finally:
             # Clear params after the completion is done
             self.params = {}
+    
+    @llm_chat_callback()
+    def chat(
+        self,
+        messages: Sequence[ChatMessage],
+        **kwargs: Any,
+    ) -> ChatResponse:
+        """ Perform a chat completion request. """
+        try:
+            logger.info(f"Sending chat request to {LLM_INFERENCE_URL} with messages: {messages}, and args: {kwargs}")
+            return self.run_async_coroutine(self.achat(messages, **kwargs))
+        except HTTPException as http_exc:
+            logger.error(f"HTTP exception during chat(): {http_exc.detail}")
+            raise http_exc
+        except Exception as e:
+            logger.error(f"Unexpected exception in chat(): {e}")
+            raise HTTPException(status_code=500, detail=f"An unexpected error occurred: {str(e)}")
 
-    async def _async_openai_complete(self, prompt: str, **kwargs: Any) -> CompletionResponse:
-        return await OpenAI(api_key=LLM_ACCESS_SECRET, **kwargs).acomplete(prompt)
+    @llm_chat_callback()
+    async def achat(
+        self,
+        messages: Sequence[ChatMessage],
+        **kwargs: Any,
+    ) -> ChatResponse:
+        """ Perform an asynchronous chat completion request. """
+        try:
+            base_model, base_max_len = self._get_default_model_info()
+            req = {
+                "model": self.get_param("model", base_model),
+                "max_tokens": self.get_param("max_tokens", base_max_len),
+                "messages": [{
+                    "role": message.role,
+                    "content": message.content if isinstance(message.content, str) else json.dumps(message.content)
+                } for message in messages if message.content is not None and message.content != ""
+                ],
+            }
 
-    async def _async_huggingface_remote_complete(self, prompt: str, **kwargs: Any) -> CompletionResponse:
-        data = {"prompt": prompt, **kwargs}
-        return await self._async_post_request(data, headers={"Authorization": f"Bearer {LLM_ACCESS_SECRET}", "Content-Type": "application/json"})
-    async def _async_custom_api_complete(self, prompt: str, **kwargs: Any) -> CompletionResponse:
+            # Add any additional parameters from self.params onto request
+            for key, value in self.params.items():
+                if key not in req:
+                    req[key] = value
+            
+            resp =  await self._async_post_request_raw(data=req, headers=DEFAULT_HEADERS)
+            return ChatResponse(
+                logprobs=resp.get("logprobs", None),
+                delta=resp.get("delta", None),
+                raw=resp,
+                message=ChatMessage(content=resp.get("choices", [{}])[0].get("message", {}).get("content", ""))
+            )
+        except HTTPException as http_exc:
+            logger.error(f"HTTP exception during achat(): {http_exc.detail}")
+            raise http_exc
+        except Exception as e:
+            logger.error(f"Unexpected exception in achat(): {e}")
+            raise HTTPException(status_code=500, detail=f"An unexpected error occurred: {str(e)}")
+        finally:
+            # Clear params after the completion is done
+            self.params = {}
+    
+    async def chat_completions_passthrough(self, chatCompletionsRequest: CompletionCreateParams, **kwargs: Any) -> ChatCompletionResponse:
+        try:
+            if "/chat/completions" not in LLM_INFERENCE_URL:
+                # If the URL does not support chat completions, raise an error
+                raise HTTPException(status_code=400, detail=f"Chat completions not supported through endpoint {LLM_INFERENCE_URL}.")
+
+            client = await self._get_httpx_client()
+            response = await client.post(LLM_INFERENCE_URL, json=chatCompletionsRequest, headers=DEFAULT_HEADERS)
+            response.raise_for_status()  # Raise an exception for HTTP errors
+            response_data = response.json()
+            # Convert to ChatCompletionResponse with source_nodes=None for passthrough
+            return ChatCompletionResponse(**response_data, source_nodes=None)
+        except HTTPException as http_exc:
+            logger.error(f"HTTP exception during chat completions passthrough: {http_exc.detail}")
+            raise http_exc
+        except httpx.HTTPStatusError as e:
+            logger.error(f"HTTP error {e.response.status_code} during POST request to {LLM_INFERENCE_URL}: {e.response.text}")
+            raise HTTPException(status_code=e.response.status_code, detail=f"{str(e.response.content)}")
+        except httpx.RequestError as e:
+            logger.error(f"Error during POST request to {LLM_INFERENCE_URL}: {e}")
+            raise HTTPException(status_code=500, detail=f"Error during POST request: {str(e)}")
+        except Exception as e:
+            logger.error(f"Unexpected error during POST request: {e}")
+            raise HTTPException(status_code=500, detail=f"Error during POST request: {str(e)}")
+
+    async def _async_completions(self, prompt: str, **kwargs: Any) -> CompletionResponse:
         model_name, model_max_len = self._get_default_model_info()
         if kwargs.get("model"):
             model_name = kwargs.pop("model")
@@ -124,24 +201,39 @@ class Inference(CustomLLM):
         # DEBUG: Call the debugging function
         # self._debug_curl_command(data)
         try:
-            return await self._async_post_request(data, headers=DEFAULT_HEADERS)
+            # add extended params
+            for key, value in self.params.items():
+                if key not in data:
+                    data[key] = value
+            resp = await self._async_post_request_raw(data, headers=DEFAULT_HEADERS)
+            return self._completions_json_to_response(resp)
         except HTTPError as e:
             if not model_name and e.response.status_code == 400:
                 logger.warning(
                     f"Potential issue with 'model' parameter in API response. "
                     f"Response: {str(e)}. Attempting to update the model name as a mitigation..."
                 )
-                self._default_model = self._fetch_default_model()  # Fetch default model dynamically
+                self._default_model, self._default_max_model_len = self._fetch_default_model_info()  # Fetch default model dynamically
                 if self._default_model:
                     logger.info(f"Default model '{self._default_model}' fetched successfully. Retrying request...")
                     data["model"] = self._default_model
-                    return await self._async_post_request(data, headers=DEFAULT_HEADERS)
+                    resp = await self._async_post_request_raw(data, headers=DEFAULT_HEADERS)
+                    return self._completions_json_to_response(resp)
                 else:
                     logger.error("Failed to fetch a default model. Aborting retry.")
             raise  # Re-raise the exception if not recoverable
         except Exception as e:
             logger.error(f"An unexpected error occurred: {e}")
             raise
+    
+    def _completions_json_to_response(self, response_json: dict) -> CompletionResponse:
+        """
+        Converts the JSON response from the completions API to a CompletionResponse object.
+        """
+        # Check if the response contains OAI format
+        if "choices" in response_json and response_json["choices"]:
+            return CompletionResponse(text=response_json["choices"][0].get("text", ""))
+        return CompletionResponse(text=str(response_json))
 
     def _get_models_endpoint(self) -> str:
         """
@@ -177,19 +269,12 @@ class Inference(CustomLLM):
             self._default_model, self._default_max_model_len = self._fetch_default_model_info()
         return self._default_model, self._default_max_model_len
 
-    async def _async_post_request(self, data: dict, headers: dict) -> CompletionResponse:
+    async def _async_post_request_raw(self, data: dict, headers: dict):
         try:
             client = await self._get_httpx_client()
             response = await client.post(LLM_INFERENCE_URL, json=data, headers=headers)
             response.raise_for_status()  # Raise an exception for HTTP errors
-            response_data = response.json()
-            # Check if the request was successful
-            if response.status_code == DEFAULT_HTTP_SUCCESS_CODE:
-                # OAI Spec returns array of choices, we choose the first one
-                if "choices" in response_data and response_data["choices"]:
-                    return CompletionResponse(text=response_data["choices"][0].get("text", ""))
-            # Return full response as text if no specific parsing is possible
-            return CompletionResponse(text=str(response_data))
+            return response.json()
         except httpx.HTTPStatusError as e:
             logger.error(f"HTTP error {e.response.status_code} during POST request to {LLM_INFERENCE_URL}: {e.response.text}")
             raise
@@ -220,7 +305,12 @@ class Inference(CustomLLM):
     @property
     def metadata(self) -> LLMMetadata:
         """Get LLM metadata."""
-        return LLMMetadata()
+        parsed_url = urlparse(LLM_INFERENCE_URL)
+        path = parsed_url.path.lower()
+        return LLMMetadata(
+            is_chat_model="/chat/completions" in path,
+            context_window=LLM_CONTEXT_WINDOW,
+        )
 
     async def aclose(self):
         """ Closes the HTTP client when shutting down. """
diff --git a/presets/ragengine/main.py b/presets/ragengine/main.py
index 382d0e0..2140c85 100644
--- a/presets/ragengine/main.py
+++ b/presets/ragengine/main.py
@@ -14,13 +14,15 @@
 
 from typing import List, Optional
 import json
+import time
 from vector_store_manager.manager import VectorStoreManager
 from embedding.huggingface_local_embedding import LocalHuggingFaceEmbedding
 from embedding.remote_embedding import RemoteEmbeddingModel
 from fastapi import FastAPI, HTTPException, Query, Request
 from models import (IndexRequest, ListDocumentsResponse, UpdateDocumentRequest,
                     QueryRequest, QueryResponse, Document, HealthStatus, DeleteDocumentRequest,
-                    DeleteDocumentResponse, UpdateDocumentResponse)
+                    DeleteDocumentResponse, UpdateDocumentResponse, ChatCompletionResponse)
+
 from vector_store.faiss_store import FaissVectorStoreHandler
 
 from ragengine.config import (REMOTE_EMBEDDING_URL, REMOTE_EMBEDDING_ACCESS_SECRET,
@@ -34,6 +36,8 @@ from starlette.responses import Response
 from ragengine.metrics.prometheus_metrics import (
     rag_query_latency,
     rag_query_requests_total,
+    rag_chat_latency,
+    rag_chat_requests_total,
     rag_index_latency,
     rag_index_requests_total,
     rag_indexes_latency,
@@ -64,8 +68,8 @@ from ragengine.metrics.prometheus_metrics import (
 app = FastAPI()
 @app.middleware("http")
 async def track_requests(request: Request, call_next):
-    tracked_paths = ["/query", "/index", "/indexes", "/persist", "/load"]
-    
+    tracked_paths = ["/query", "/index", "/indexes", "/persist", "/load", "/v1/chat/completions"]
+
     should_track = any(request.url.path.startswith(path) for path in tracked_paths)
     
     if not should_track:
@@ -278,6 +282,77 @@ async def query_index(request: QueryRequest):
         rag_query_requests_total.labels(status=status).inc()
         rag_query_latency.labels(status=status).observe(time.perf_counter() - start_time)
 
+@app.post(
+    "/v1/chat/completions",
+    response_model=ChatCompletionResponse,
+    summary="OpenAI-Compatible Chat Completions API",
+    description="""
+    OpenAI-compatible chat completions endpoint with RAG capabilities.
+
+    ## Request Example:
+    ```json
+    {
+      "index_name": "example_index",
+      "model": "example_model",
+      "messages": [
+        {"role": "system", "content": "You are a knowledgeable assistant."},
+        {"role": "user", "content": "What is RAG?"}
+      ],
+      "temperature": 0.7,
+      "max_tokens": 2048,
+      "top_k": 5,
+      "rerank_params": {"top_n": 3}
+    }
+    ```
+
+    ## Response Example:
+    ```json
+    {
+      "id": "chatcmpl-123",
+      "object": "chat.completion",
+      "created": 1677652288,
+      "model": "example_model",
+      "choices": [
+        {
+          "index": 0,
+          "message": {
+            "role": "assistant",
+            "content": "RAG stands for Retrieval-Augmented Generation..."
+          },
+          "finish_reason": "stop"
+        }
+      ],
+      "usage": {
+        "prompt_tokens": 56,
+        "completion_tokens": 31,
+        "total_tokens": 87
+      },
+      "source_nodes": [...]
+    }
+    ```
+    """,
+)
+async def chat_completions(request: dict):
+    start_time = time.perf_counter()
+    status = STATUS_FAILURE  # Default status
+    try:
+        response = await rag_ops.chat_completion(request)
+        status = STATUS_SUCCESS
+        return response
+    except HTTPException as http_exc:
+        # Preserve HTTP exceptions like 422 from reranker
+        raise http_exc
+    except ValueError as ve:
+        raise HTTPException(status_code=400, detail=str(ve))  # Validation issue
+    except Exception as e:
+        raise HTTPException(
+            status_code=500, detail=f"An unexpected error occurred: {str(e)}"
+        )
+    finally:
+        # Record metrics once in finally block
+        rag_chat_requests_total.labels(status=status).inc()
+        rag_chat_latency.labels(status=status).observe(time.perf_counter() - start_time)
+
 @app.get(
     "/indexes",
     response_model=List[str],
diff --git a/presets/ragengine/metrics/prometheus_metrics.py b/presets/ragengine/metrics/prometheus_metrics.py
index 353b8a2..51bf689 100644
--- a/presets/ragengine/metrics/prometheus_metrics.py
+++ b/presets/ragengine/metrics/prometheus_metrics.py
@@ -39,6 +39,10 @@ rag_embedding_requests_total = Counter(
 rag_query_latency = Histogram('rag_query_latency_seconds', 'Time to call \'/query\' API in seconds', labelnames=[STATUS_LABEL])
 rag_query_requests_total = Counter('rag_query_requests_total', 'Count of successful/failed calling \'/query\' requests', labelnames=[STATUS_LABEL])
 
+# Chat API metrics
+rag_chat_latency = Histogram('rag_chat_latency_seconds', 'Time to call \'/v1/chat/completions\' API in seconds', labelnames=[STATUS_LABEL])
+rag_chat_requests_total = Counter('rag_chat_requests_total', 'Count of successful/failed calling \'/v1/chat/completions\' requests', labelnames=[STATUS_LABEL])
+
 # Index API metrics
 rag_index_latency = Histogram('rag_index_latency_seconds', 'Time to call \'/index\' API in seconds', labelnames=[STATUS_LABEL])
 rag_index_requests_total = Counter('rag_index_requests_total', 'Count of successful/failed calling \'/index\' requests', labelnames=[STATUS_LABEL])
diff --git a/presets/ragengine/models.py b/presets/ragengine/models.py
index 3a35ff5..7490b00 100644
--- a/presets/ragengine/models.py
+++ b/presets/ragengine/models.py
@@ -13,8 +13,18 @@
 
 
 from typing import Any, Dict, List, Optional
+from typing_extensions import Literal, Required, Optional
+from openai.types.chat import (
+    ChatCompletion,
+    CompletionCreateParams,
+    ChatCompletionMessageParam,
+    ChatCompletionUserMessageParam,
+    ChatCompletionSystemMessageParam,
+    ChatCompletionDeveloperMessageParam,
+    ChatCompletionContentPartTextParam,
+)
 
-from pydantic import BaseModel, Field, model_validator
+from pydantic import BaseModel, Field, model_validator, ValidationError, create_model
 
 
 class Document(BaseModel):
@@ -93,4 +103,36 @@ class QueryResponse(BaseModel):
 
 class HealthStatus(BaseModel):
     status: str
-    detail: Optional[str] = None 
+    detail: Optional[str] = None
+
+class ChatCompletionResponse(ChatCompletion):
+    source_nodes: Optional[List[NodeWithScore]] = None
+
+def messages_to_prompt(messages: List[Dict]) -> str:
+    """Convert messages to a prompt string."""
+    string_messages = []
+    for message in messages:
+        content = get_message_content(message)
+        string_messages.append(f"{message.get('role')}: {content}")
+    return "\n".join(string_messages)
+
+def get_message_content(message: Dict) -> str:
+    """Extract content from a ChatCompletionMessageParam."""
+    if message.get("role") == "user":
+        if message.get("content"):
+            content = message.get("content")
+            if isinstance(content, str):
+                return content
+            elif isinstance(content, dict) and content.get("type") == "text":
+                return content.get("text", "")
+            elif isinstance(content, list):
+                user_text_content = []
+                for part in content:
+                    if isinstance(part, str):
+                        user_text_content.append(part)
+                    elif part.get("type") == "text":
+                        user_text_content.append(part.get("text", ""))
+                return "\n".join(user_text_content)
+        return ""
+    else:
+        return message.get("content", {}).get("text", "") if isinstance(message.get("content"), dict) else message.get("content", "")
\ No newline at end of file
diff --git a/presets/ragengine/requirements.txt b/presets/ragengine/requirements.txt
index a6007e9..b6c513a 100644
--- a/presets/ragengine/requirements.txt
+++ b/presets/ragengine/requirements.txt
@@ -16,6 +16,7 @@ uvicorn==0.34.2
 asyncio==3.4.3
 aiorwlock==1.5.0
 httpx==0.27.0
+openai==1.97.1
 
 tree-sitter==0.23.2
 tree_sitter_languages==1.10.2
diff --git a/presets/ragengine/vector_store/base.py b/presets/ragengine/vector_store/base.py
index 8abfacf..7cb4b8c 100644
--- a/presets/ragengine/vector_store/base.py
+++ b/presets/ragengine/vector_store/base.py
@@ -15,18 +15,23 @@
 import logging
 from abc import ABC, abstractmethod
 from typing import Dict, List, Optional, Any
+from pydantic import ValidationError
 import hashlib
 import os
 import asyncio
 from itertools import islice
+import uuid
+import time
 
 from llama_index.core import Document as LlamaDocument
 from llama_index.core import (StorageContext, VectorStoreIndex, load_index_from_storage)
+from llama_index.core.chat_engine.types import ChatMode
 from llama_index.core.postprocessor import LLMRerank  # Query with LLM Reranking
 
 from llama_index.vector_stores.faiss import FaissMapVectorStore
+from openai.types.chat import ChatCompletionContentPartTextParam, CompletionCreateParams
 
-from ragengine.models import Document
+from ragengine.models import Document, ChatCompletionResponse, messages_to_prompt
 from ragengine.embedding.base import BaseEmbeddingModel
 from ragengine.inference.inference import Inference
 from ragengine.vector_store.transformers.custom_transformer import CustomTransformer
@@ -156,8 +161,6 @@ class BaseVectorStore(ABC):
         if index_name not in self.index_map:
             raise HTTPException(status_code=404, detail=f"No such index: '{index_name}' exists.")
 
-        self.llm.set_params(llm_params)
-
         node_postprocessors = []
         if rerank_params:
             # Set default reranking parameters and merge with provided params
@@ -176,16 +179,21 @@ class BaseVectorStore(ABC):
                 )
             )
 
-        query_engine = self.index_map[index_name].as_query_engine(
+        query_engine = self.index_map[index_name].as_chat_engine(
             llm=self.llm,
             similarity_top_k=top_k,
-            node_postprocessors=node_postprocessors
+            node_postprocessors=node_postprocessors,
+            chat_mode=ChatMode.CONDENSE_PLUS_CONTEXT,
+            verbose=True
         )
+
         if self.use_rwlock:
             async with self.rwlock.reader_lock:
-                query_result = await query_engine.aquery(query)
+                self.llm.set_params(llm_params)
+                query_result = await query_engine.achat(query)
         else:
-            query_result = await query_engine.aquery(query)
+            self.llm.set_params(llm_params)
+            query_result = await query_engine.achat(query)
         return {
             "response": query_result.response,
             "source_nodes": [
@@ -201,6 +209,134 @@ class BaseVectorStore(ABC):
             "metadata": query_result.metadata,
         }
 
+    async def chat_completion(self, request: dict) -> ChatCompletionResponse:
+        """
+        Create a chat completion based on the provided request.
+
+        Args:
+            request (ChatCompletionRequest): The request containing the chat messages and parameters.
+
+        Returns:
+            ChatCompletionResponse: The response containing the generated chat completion.
+        """
+        if request.get("index_name") and request.get("index_name") not in self.index_map:
+            raise HTTPException(status_code=404, detail=f"No such index: '{request.get('index_name')}' exists.")
+
+        llm_params = {}
+        if request.get("model") is not None:
+            llm_params["model"] = request.get("model")
+        if request.get("temperature") is not None:
+            llm_params["temperature"] = request.get("temperature")
+        if request.get("top_p") is not None:
+            llm_params["top_p"] = request.get("top_p")
+        if request.get("max_tokens") is not None:
+            llm_params["max_tokens"] = request.get("max_tokens")
+
+        logger.info("converting request to OpenAI format")
+        openai_request = None
+        last_error = None
+        for model_cls in CompletionCreateParams.__args__:
+            try:
+                openai_request = model_cls(**request)
+                break
+            except ValidationError as e:
+                last_error = e
+
+        if openai_request is None:
+            logger.error(f"Invalid request format: {str(last_error)}")
+            raise HTTPException(status_code=400, detail=f"Invalid request format: {str(last_error)}")
+        
+        if not request.get("index_name"):
+            logger.info(f"Request does not specify an index, passing through to LLM directly.")
+            return await self.llm.chat_completions_passthrough(openai_request)
+
+        if request.get("tools") or request.get("functions"):
+            logger.info(f"Request contains tools or functions, passing through to LLM directly.")
+            return await self.llm.chat_completions_passthrough(openai_request)
+
+        # Only support RAG usage on user/system/developer roles in messages and only text content
+        for message in request.get("messages", []):
+            # Every message must have a role
+            if not message.get("role"):
+                raise HTTPException(status_code=400, detail=f"Invalid request format: messages must contain 'role'.")
+            
+            # Every message must have content aside from assistant role messages
+            if message.get("role") != "assistant" and message.get("content") is None:
+                raise HTTPException(status_code=400, detail=f"Invalid request format: messages must contain 'content' for role '{message.get('role')}'.")
+
+            # Only user, system, and developer roles are supported for RAG
+            if message.get("role") not in ["user", "system", "assistant", "developer"]:
+                logger.info(f"Request contains unsupported role '{message.get('role')}' in messages, passing through to LLM directly.")
+                return await self.llm.chat_completions_passthrough(openai_request)
+
+            # User message content can be a range of options, but we only support text content for RAG
+            if message.get("role") == "user":
+                if message.get("content"):
+                    content = message.get("content")
+                    if isinstance(content, list):
+                        for part in content:
+                            if not isinstance(part, str) and part.get("type") != "text":
+                                logger.info(f"Request contains unsupported content type in user message, passing through to LLM directly.")
+                                return await self.llm.chat_completions_passthrough(openai_request)
+                    elif isinstance(content, str):
+                        pass
+                    elif isinstance(content, ChatCompletionContentPartTextParam):
+                        pass
+                    else:
+                        logger.info(f"Request contains unsupported content type '{type(content)}' in messages, passing through to LLM directly.")
+                        return await self.llm.chat_completions_passthrough(openai_request)
+                else:
+                    logger.error(f"Invalid request format: user messages must contain 'content'.")
+                    raise HTTPException(status_code=400, detail=f"Invalid request format: user messages must contain 'content'.")
+
+        prompt = messages_to_prompt(request.get("messages", []))
+
+        logger.info(f"Creating chat engine for index '{request.get('index_name')}' with prompt: {prompt}")
+        chat_engine = self.index_map[request.get("index_name")].as_chat_engine(
+            llm=self.llm,
+            similarity_top_k=request.get("top_k", 5),
+            chat_mode=ChatMode.CONDENSE_PLUS_CONTEXT,
+            verbose=True
+        )
+
+        logger.info("Processing chat completion request with prompt.")
+        try:
+            if self.use_rwlock:
+                async with self.rwlock.reader_lock:
+                    self.llm.set_params(llm_params)
+                    chat_result = await chat_engine.achat(prompt)
+            else:
+                self.llm.set_params(llm_params)
+                chat_result = await chat_engine.achat(prompt)
+
+            return ChatCompletionResponse(
+                id=uuid.uuid4().hex,
+                object="chat.completion",
+                created=int(time.time()),
+                model=request.get("model"),
+                choices=[{
+                    "message": {
+                        "role": "assistant",
+                        "content": chat_result.response,
+                    },
+                    "finish_reason": "stop",
+                    "index": 0
+                }],
+                source_nodes=[
+                    {
+                        "doc_id": source_node.node.ref_doc_id,
+                        "node_id": source_node.node_id,
+                        "text": source_node.text,
+                        "score": source_node.score,
+                        "metadata": source_node.metadata
+                    }
+                    for source_node in chat_result.source_nodes
+                ],
+            )
+        except Exception as e:
+            logger.error(f"Error during chat completion: {str(e)}")
+            raise HTTPException(status_code=500, detail=f"Chat completion failed: {str(e)}")
+
     async def add_document_to_index(self, index_name: str, document: Document, doc_id: str):
         """Common logic for adding a single document."""
         if index_name not in self.index_map:
diff --git a/presets/ragengine/vector_store_manager/manager.py b/presets/ragengine/vector_store_manager/manager.py
index d1e4350..84bdb14 100644
--- a/presets/ragengine/vector_store_manager/manager.py
+++ b/presets/ragengine/vector_store_manager/manager.py
@@ -17,6 +17,7 @@ from typing import Dict, List, Any
 from ragengine.models import Document
 from ragengine.vector_store.base import BaseVectorStore
 
+
 class VectorStoreManager:
     def __init__(self, vector_store: BaseVectorStore):
         self.vector_store = vector_store
@@ -35,6 +36,10 @@ class VectorStoreManager:
         """Query the indexed documents."""
         return await self.vector_store.query(index_name, query, top_k, llm_params, rerank_params)
 
+    async def chat_completion(self, request: dict):
+        """Chat completion using the vector store."""
+        return await self.vector_store.chat_completion(request)
+
     def list_indexes(self):
         """List all indexes."""
         return self.vector_store.list_indexes()



Test patch:
diff --git a/presets/ragengine/tests/api/conftest.py b/presets/ragengine/tests/api/conftest.py
index 6af7838..5f20223 100644
--- a/presets/ragengine/tests/api/conftest.py
+++ b/presets/ragengine/tests/api/conftest.py
@@ -20,4 +20,24 @@ os.environ["OMP_NUM_THREADS"] = "1" # Force single-threaded for testing to preve
 os.environ["MKL_NUM_THREADS"] = "1"  # Force MKL to use a single thread
 
 import nest_asyncio
+import asyncio
+import pytest_asyncio
+import httpx
+from ragengine.main import app, vector_store_handler
 nest_asyncio.apply()
+
+@pytest_asyncio.fixture
+async def async_client():
+    """Use an async HTTP client to interact with FastAPI app."""
+    async with httpx.AsyncClient(app=app, base_url="http://localhost") as client:
+        yield client
+
+@pytest_asyncio.fixture(scope="session")
+def event_loop():
+    loop = asyncio.new_event_loop()
+    yield loop
+    loop.close()
+
+@pytest_asyncio.fixture(autouse=True)
+def clear_index():
+    vector_store_handler.index_map.clear()
\ No newline at end of file
diff --git a/presets/ragengine/tests/api/test_chat_completions.py b/presets/ragengine/tests/api/test_chat_completions.py
new file mode 100644
index 0000000..2e53806
--- /dev/null
+++ b/presets/ragengine/tests/api/test_chat_completions.py
@@ -0,0 +1,800 @@
+# Copyright (c) KAITO authors.
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from unittest.mock import patch
+
+import pytest
+import httpx
+import respx
+import time
+import re
+
+@pytest.fixture(autouse=True)
+def overwrite_inference_url(monkeypatch):
+    import ragengine.inference.inference
+    import ragengine.config
+    monkeypatch.setattr(ragengine.config, "LLM_INFERENCE_URL", "http://localhost:5000/v1/chat/completions")
+    monkeypatch.setattr(ragengine.inference.inference, "LLM_INFERENCE_URL", "http://localhost:5000/v1/chat/completions")
+
+@pytest.mark.asyncio
+@respx.mock
+@patch("requests.get")
+async def test_chat_completions_basic_success(mock_get, async_client):
+    """Test basic successful chat completion with RAG functionality."""
+    # Mock the response for the default model fetch
+    mock_get.return_value.status_code = 200
+    mock_get.return_value.json.return_value = {
+        "data": [{"id": "mock-model", "max_model_len": 2048}]
+    }
+
+    # Mock HTTPX response for Custom Inference API
+    mock_response = {
+        "id": "chatcmpl-test123",
+        "object": "chat.completion",
+        "created": int(time.time()),
+        "model": "mock-model",
+        "choices": [
+            {
+                "index": 0,
+                "message": {
+                    "role": "assistant",
+                    "content": "This is a helpful response about the test document."
+                },
+                "finish_reason": "stop"
+            }
+        ],
+        "usage": {
+            "prompt_tokens": 25,
+            "completion_tokens": 12,
+            "total_tokens": 37
+        }
+    }
+    respx.post("http://localhost:5000/v1/chat/completions").mock(return_value=httpx.Response(200, json=mock_response))
+
+    # Index some test documents
+    index_request = {
+        "index_name": "test_index",
+        "documents": [
+            {"text": "This is a test document about AI and machine learning."},
+            {"text": "Another document discussing natural language processing."}
+        ]
+    }
+
+    response = await async_client.post("/index", json=index_request)
+    assert response.status_code == 200
+
+    # Test chat completion request
+    chat_request = {
+        "index_name": "test_index",
+        "model": "mock-model",
+        "messages": [
+            {"role": "user", "content": "What can you tell me about AI?"}
+        ],
+        "temperature": 0.7,
+        "max_tokens": 100,
+        "top_k": 2
+    }
+
+    response = await async_client.post("/v1/chat/completions", json=chat_request)
+    assert response.status_code == 200
+    
+    response_data = response.json()
+    assert "id" in response_data
+    assert response_data["object"] == "chat.completion"
+    assert "created" in response_data
+    assert response_data["model"] == "mock-model"
+    assert len(response_data["choices"]) == 1
+    assert response_data["choices"][0]["message"]["role"] == "assistant"
+    assert response_data["choices"][0]["message"]["content"] == "This is a helpful response about the test document."
+    assert response_data["choices"][0]["finish_reason"] == "stop"
+    assert response_data["choices"][0]["index"] == 0
+    assert "source_nodes" in response_data
+    assert len(response_data["source_nodes"]) > 0
+
+    response = await async_client.get(f"/metrics")
+    assert response.status_code == 200
+    assert len(re.findall(r'rag_index_requests_total{status="success"} ([1-9]\d*).0', response.text)) == 1
+    assert len(re.findall(r'rag_chat_requests_total{status="success"} ([1-9]\d*).0', response.text)) == 1
+
+@pytest.mark.asyncio
+@respx.mock
+@patch("requests.get")
+async def test_chat_completions_without_index_name(mock_get, async_client):
+    """Test chat completion request without index_name (should passthrough to LLM)."""
+    # Mock the response for the default model fetch
+    mock_get.return_value.status_code = 200
+    mock_get.return_value.json.return_value = {
+        "data": [{"id": "mock-model", "max_model_len": 2048}]
+    }
+
+    # Mock HTTPX response for passthrough LLM call
+    mock_response = {
+        "id": "chatcmpl-test123",
+        "object": "chat.completion",
+        "created": int(time.time()),
+        "model": "mock-model",
+        "choices": [{
+            "index": 0,
+            "message": {"role": "assistant", "content": "This is a direct LLM response"},
+            "finish_reason": "stop"
+        }]
+    }
+    respx.post("http://localhost:5000/v1/chat/completions").mock(return_value=httpx.Response(200, json=mock_response))
+
+    # Test request without index_name (should trigger passthrough)
+    chat_request = {
+        "model": "mock-model",
+        "messages": [
+            {"role": "user", "content": "Hello, how are you?"}
+        ],
+        "temperature": 0.7,
+        "max_tokens": 100
+    }
+
+    response = await async_client.post("/v1/chat/completions", json=chat_request)
+    assert response.status_code == 200
+    
+    response_data = response.json()
+    assert response_data["id"] == "chatcmpl-test123"
+    assert response_data["choices"][0]["message"]["content"] == "This is a direct LLM response"
+    # Should have source_nodes field but it should be None for passthrough requests
+    assert response_data["source_nodes"] is None
+
+@pytest.mark.asyncio
+@respx.mock
+@patch("requests.get")
+async def test_chat_completions_with_tools(mock_get, async_client):
+    """Test chat completion with tools (should passthrough to LLM)."""
+    # Mock the response for the default model fetch
+    mock_get.return_value.status_code = 200
+    mock_get.return_value.json.return_value = {
+        "data": [{"id": "mock-model", "max_model_len": 2048}]
+    }
+
+    # Mock HTTPX response for passthrough LLM call
+    mock_response = {
+        "id": "chatcmpl-tools123",
+        "object": "chat.completion",
+        "created": int(time.time()),
+        "model": "mock-model",
+        "choices": [{
+            "index": 0,
+            "message": {
+                "role": "assistant", 
+                "content": None,
+                "tool_calls": [{"id": "call123", "type": "function", "function": {"name": "test_tool", "arguments": '{"param1": "value1"}'}}]
+            },
+            "finish_reason": "tool_calls"
+        }]
+    }
+    respx.post("http://localhost:5000/v1/chat/completions").mock(return_value=httpx.Response(200, json=mock_response))
+
+    # Test request with tools (should trigger passthrough)
+    chat_request = {
+        "model": "mock-model",
+        "messages": [
+            {"role": "user", "content": "Use a tool to help me"}
+        ],
+        "tools": [
+            {
+                "type": "function",
+                "function": {
+                    "name": "test_tool",
+                    "description": "A test tool",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "param1": {"type": "string", "description": "A test parameter"}
+                        },
+                    }
+                }
+            }
+        ]
+    }
+
+    response = await async_client.post("/v1/chat/completions", json=chat_request)
+    assert response.status_code == 200
+    
+    response_data = response.json()
+    assert response_data["choices"][0]["finish_reason"] == "tool_calls"
+    assert "tool_calls" in response_data["choices"][0]["message"]
+
+@pytest.mark.asyncio
+async def test_chat_completions_nonexistent_index(async_client):
+    """Test chat completion with non-existent index."""
+    chat_request = {
+        "index_name": "nonexistent_index",
+        "model": "mock-model",
+        "messages": [
+            {"role": "user", "content": "Test question"}
+        ]
+    }
+
+    response = await async_client.post("/v1/chat/completions", json=chat_request)
+    assert response.status_code == 404
+    assert "No such index: 'nonexistent_index' exists" in response.json()["detail"]
+
+@pytest.mark.asyncio
+@respx.mock
+@patch("requests.get")
+async def test_chat_completions_invalid_request_format(mock_get, async_client):
+    """Test chat completion with invalid request format."""
+    # Mock the response for the default model fetch
+    mock_get.return_value.status_code = 200
+    mock_get.return_value.json.return_value = {
+        "data": [{"id": "mock-model", "max_model_len": 2048}]
+    }
+
+    # Mock HTTPX response for passthrough LLM call (in case it gets that far)
+    respx.post("http://localhost:5000/v1/chat/completions").mock(return_value=httpx.Response(400, json={"error": "Invalid request"}))
+
+    # Test missing messages
+    chat_request = {
+        "model": "mock-model"
+    }
+
+    response = await async_client.post("/v1/chat/completions", json=chat_request)
+    assert response.status_code == 400
+    assert "Invalid request" in response.json()["detail"]
+
+@pytest.mark.asyncio
+async def test_chat_completions_missing_role_in_message(async_client):
+    """Test chat completion with message missing role."""
+    # Index some test documents
+    index_request = {
+        "index_name": "test_index",
+        "documents": [
+            {"text": "This is a test document about AI and machine learning."},
+            {"text": "Another document discussing natural language processing."}
+        ]
+    }
+
+    response = await async_client.post("/index", json=index_request)
+    assert response.status_code == 200
+
+    chat_request = {
+        "index_name": "test_index",
+        "model": "mock-model",
+        "messages": [
+            {"content": "What can you tell me about AI?"}
+        ],
+        "temperature": 0.7,
+        "max_tokens": 100,
+        "top_k": 2
+    }
+
+    response = await async_client.post("/v1/chat/completions", json=chat_request)
+    assert response.status_code == 400
+    assert "messages must contain 'role'" in response.json()["detail"]
+
+@pytest.mark.asyncio
+async def test_chat_completions_missing_content_for_user_role(async_client):
+    """Test chat completion with user message missing content."""
+    # Index some test documents
+    index_request = {
+        "index_name": "test_index",
+        "documents": [
+            {"text": "This is a test document about AI and machine learning."},
+            {"text": "Another document discussing natural language processing."}
+        ]
+    }
+
+    response = await async_client.post("/index", json=index_request)
+    assert response.status_code == 200
+
+    chat_request = {
+        "index_name": "test_index",
+        "model": "mock-model",
+        "messages": [
+            {"role": "user"}
+        ],
+        "temperature": 0.7,
+        "max_tokens": 100,
+        "top_k": 2
+    }
+
+    response = await async_client.post("/v1/chat/completions", json=chat_request)
+    assert response.status_code == 400
+    assert "messages must contain 'content' for role 'user'" in response.json()["detail"]
+
+@pytest.mark.asyncio
+@respx.mock
+@patch("requests.get")
+async def test_chat_completions_system_message(mock_get, async_client):
+    """Test chat completion with system message."""
+    # Mock the response for the default model fetch
+    mock_get.return_value.status_code = 200
+    mock_get.return_value.json.return_value = {
+        "data": [{"id": "mock-model", "max_model_len": 2048}]
+    }
+
+    # Mock HTTPX response for Custom Inference API
+    mock_response = {
+        "id": "chatcmpl-test123",
+        "object": "chat.completion",
+        "created": int(time.time()),
+        "model": "mock-model",
+        "choices": [
+            {
+                "index": 0,
+                "message": {
+                    "role": "assistant",
+                    "content": "This is a helpful response about the test document."
+                },
+                "finish_reason": "stop"
+            }
+        ],
+        "usage": {
+            "prompt_tokens": 25,
+            "completion_tokens": 12,
+            "total_tokens": 37
+        }
+    }
+    respx.post("http://localhost:5000/v1/chat/completions").mock(return_value=httpx.Response(200, json=mock_response))
+
+    # Index some test documents
+    index_request = {
+        "index_name": "test_index",
+        "documents": [
+            {"text": "Document about machine learning algorithms."}
+        ]
+    }
+
+    response = await async_client.post("/index", json=index_request)
+    assert response.status_code == 200
+
+    # Test chat completion with system message
+    chat_request = {
+        "index_name": "test_index",
+        "model": "mock-model",
+        "messages": [
+            {"role": "system", "content": "You are a helpful AI assistant specializing in machine learning."},
+            {"role": "user", "content": "Tell me about algorithms."}
+        ],
+        "temperature": 0.5
+    }
+
+    response = await async_client.post("/v1/chat/completions", json=chat_request)
+    assert response.status_code == 200
+    
+    response_data = response.json()
+    assert response_data["choices"][0]["message"]["content"] == "This is a helpful response about the test document."
+    assert len(response_data["source_nodes"]) > 0
+
+@pytest.mark.asyncio
+@respx.mock
+@patch("requests.get")
+async def test_chat_completions_unsupported_message_role(mock_get, async_client):
+    """Test chat completion with unsupported message role (should passthrough)."""
+    # Mock the response for the default model fetch
+    mock_get.return_value.status_code = 200
+    mock_get.return_value.json.return_value = {
+        "data": [{"id": "mock-model", "max_model_len": 2048}]
+    }
+
+    # Mock HTTPX response for passthrough LLM call
+    mock_response = {"detail": "bad request format"}
+    respx.post("http://localhost:5000/v1/chat/completions").mock(return_value=httpx.Response(400, json=mock_response))
+
+    # Test request with unsupported role (should trigger passthrough)
+    chat_request = {
+        "model": "mock-model",
+        "messages": [
+            {"role": "function", "content": "Function response", "name": "test_function"}
+        ]
+    }
+
+    response = await async_client.post("/v1/chat/completions", json=chat_request)
+    assert response.status_code == 400
+    
+    response_data = response.json()
+    assert "bad request format" in response_data["detail"]
+
+@pytest.mark.asyncio
+@respx.mock
+@patch("requests.get")
+async def test_chat_completions_complex_user_content(mock_get, async_client):
+    """Test chat completion with complex user content (should passthrough)."""
+    # Mock the response for the default model fetch
+    mock_get.return_value.status_code = 200
+    mock_get.return_value.json.return_value = {
+        "data": [{"id": "mock-model", "max_model_len": 2048}]
+    }
+
+    # Mock HTTPX response for passthrough LLM call
+    mock_response = {
+        "id": "chatcmpl-complex",
+        "object": "chat.completion",
+        "created": int(time.time()),
+        "model": "mock-model",
+        "choices": [{
+            "index": 0,
+            "message": {"role": "assistant", "content": "Complex content response"},
+            "finish_reason": "stop"
+        }]
+    }
+    respx.post("http://localhost:5000/v1/chat/completions").mock(return_value=httpx.Response(200, json=mock_response))
+
+    # Test request with complex content (should trigger passthrough)
+    chat_request = {
+        "model": "mock-model",
+        "messages": [
+            {
+                "role": "user", 
+                "content": [
+                    {"type": "text", "text": "What's in this image?"},
+                    {"type": "image_url", "image_url": {"url": "data:image/jpeg;base64,..."}}
+                ]
+            }
+        ]
+    }
+
+    response = await async_client.post("/v1/chat/completions", json=chat_request)
+    assert response.status_code == 200
+    
+    response_data = response.json()
+    assert response_data["choices"][0]["message"]["content"] == "Complex content response"
+    # Should have source_nodes field but it should be None for passthrough requests
+    assert response_data["source_nodes"] is None
+
+@pytest.mark.asyncio
+@respx.mock
+@patch("requests.get")
+async def test_chat_completions_developer_role(mock_get, async_client):
+    """Test chat completion with developer role message."""
+    # Mock the response for the default model fetch
+    mock_get.return_value.status_code = 200
+    mock_get.return_value.json.return_value = {
+        "data": [{"id": "mock-model", "max_model_len": 2048}]
+    }
+
+    # Mock HTTPX response for Custom Inference API
+    mock_response = {
+        "id": "chatcmpl-test123",
+        "object": "chat.completion",
+        "model": "mock-model",
+        "choices": [
+            {
+                "index": 0,
+                "message": {
+                    "role": "assistant",
+                    "content": "This is a helpful response about the test document."
+                },
+                "finish_reason": "stop"
+            }
+        ]
+    }
+    respx.post("http://localhost:5000/v1/chat/completions").mock(return_value=httpx.Response(200, json=mock_response))
+
+    # Index some test documents
+    index_request = {
+        "index_name": "test_index",
+        "documents": [
+            {"text": "Technical documentation about APIs."}
+        ]
+    }
+
+    response = await async_client.post("/index", json=index_request)
+    assert response.status_code == 200
+
+    # Test chat completion with developer message
+    chat_request = {
+        "index_name": "test_index",
+        "model": "mock-model",
+        "messages": [
+            {"role": "developer", "content": "Debug information: API call failed"},
+            {"role": "user", "content": "Help me understand the API."}
+        ]
+    }
+
+    response = await async_client.post("/v1/chat/completions", json=chat_request)
+    assert response.status_code == 200
+    
+    response_data = response.json()
+    assert response_data["choices"][0]["message"]["content"] == "This is a helpful response about the test document."
+
+@pytest.mark.asyncio
+@respx.mock
+@patch("requests.get")
+async def test_chat_completions_with_top_k_parameter(mock_get, async_client):
+    """Test chat completion with custom top_k parameter."""
+    # Mock the response for the default model fetch
+    mock_get.return_value.status_code = 200
+    mock_get.return_value.json.return_value = {
+        "data": [{"id": "mock-model", "max_model_len": 2048}]
+    }
+
+    # Mock HTTPX response for Custom Inference API
+    mock_response = {"result": "Response based on top documents."}
+    respx.post("http://localhost:5000/v1/chat/completions").mock(return_value=httpx.Response(200, json=mock_response))
+
+    # Index multiple test documents
+    index_request = {
+        "index_name": "test_index",
+        "documents": [
+            {"text": "First document about machine learning."},
+            {"text": "Second document about deep learning."},
+            {"text": "Third document about neural networks."},
+            {"text": "Fourth document about AI applications."},
+            {"text": "Fifth document about data science."}
+        ]
+    }
+
+    response = await async_client.post("/index", json=index_request)
+    assert response.status_code == 200
+
+    # Test chat completion with custom top_k
+    chat_request = {
+        "index_name": "test_index",
+        "model": "mock-model",
+        "messages": [
+            {"role": "user", "content": "Tell me about machine learning."}
+        ],
+        "top_k": 3
+    }
+
+    response = await async_client.post("/v1/chat/completions", json=chat_request)
+    assert response.status_code == 200
+    
+    response_data = response.json()
+    # Should return up to 3 source nodes based on top_k parameter
+    assert len(response_data["source_nodes"]) <= 3
+
+@pytest.mark.asyncio
+@respx.mock
+@patch("requests.get")
+async def test_chat_completions_error_handling(mock_get, async_client):
+    """Test chat completion error handling when LLM call fails."""
+    # Mock the response for the default model fetch
+    mock_get.return_value.status_code = 200
+    mock_get.return_value.json.return_value = {
+        "data": [{"id": "mock-model", "max_model_len": 2048}]
+    }
+
+    # Mock HTTPX response with error
+    respx.post("http://localhost:5000/v1/chat/completions").mock(return_value=httpx.Response(500, json={"error": "Internal server error"}))
+
+    # Index some test documents
+    index_request = {
+        "index_name": "test_index",
+        "documents": [
+            {"text": "Test document."}
+        ]
+    }
+
+    response = await async_client.post("/index", json=index_request)
+    assert response.status_code == 200
+
+    # Test chat completion that should fail
+    chat_request = {
+        "index_name": "test_index",
+        "model": "mock-model",
+        "messages": [
+            {"role": "user", "content": "Test question."}
+        ]
+    }
+
+    response = await async_client.post("/v1/chat/completions", json=chat_request)
+    assert response.status_code == 500
+    assert "An unexpected error occurred" in response.json()["detail"]
+
+@pytest.mark.asyncio
+@respx.mock 
+@patch("requests.get")
+async def test_chat_completions_assistant_message_with_content(mock_get, async_client):
+    """Test chat completion with assistant message that has content."""
+    # Mock the response for the default model fetch
+    mock_get.return_value.status_code = 200
+    mock_get.return_value.json.return_value = {
+        "data": [{"id": "mock-model", "max_model_len": 2048}]
+    }
+
+    # Mock HTTPX response for Custom Inference API
+    mock_response = {
+            "id": "chatcmpl-test123",
+            "object": "chat.completion",
+            "model": "mock-model",
+            "choices": [
+                {
+                    "index": 0,
+                    "message": {
+                        "role": "assistant",
+                        "content": "This is a helpful response about the test document."
+                    },
+                    "finish_reason": "stop"
+                }
+            ]
+        }
+    respx.post("http://localhost:5000/v1/chat/completions").mock(return_value=httpx.Response(200, json=mock_response))
+
+    # Index some test documents
+    index_request = {
+        "index_name": "test_index",
+        "documents": [
+            {"text": "Conversation about AI."}
+        ]
+    }
+
+    response = await async_client.post("/index", json=index_request)
+    assert response.status_code == 200
+
+    # Test chat completion with assistant message without content
+    chat_request = {
+        "index_name": "test_index",
+        "model": "mock-model",
+        "messages": [
+            {"role": "user", "content": "Hello"},
+            {"role": "assistant", "content": "Hello! How can I help you?"},  # Assistant message with content
+            {"role": "user", "content": "Can you help me?"}
+        ]
+    }
+
+    response = await async_client.post("/v1/chat/completions", json=chat_request)
+    assert response.status_code == 200
+    
+    response_data = response.json()
+    assert response_data["choices"][0]["message"]["content"] == "This is a helpful response about the test document."
+
+@pytest.mark.asyncio
+async def test_chat_completions_metrics_tracking(async_client):
+    """Test that metrics are properly tracked for chat completions."""
+    # Test successful request
+    chat_request = {
+        "model": "mock-model",
+        "messages": [
+            {"role": "user", "content": "Hello"}
+        ]
+    }
+
+    # This will fail but should still track metrics
+    await async_client.post("/v1/chat/completions", json=chat_request)
+    # Should fail due to validation error or missing setup
+    
+    # Check metrics endpoint
+    metrics_response = await async_client.get("/metrics")
+    assert metrics_response.status_code == 200
+    
+    # Should have at least one chat request recorded (success or failure)
+    metrics_text = metrics_response.text
+    assert "rag_chat_requests_total" in metrics_text
+    assert "rag_chat_latency" in metrics_text
+
+@pytest.mark.asyncio 
+@respx.mock
+@patch("requests.get")
+async def test_chat_completions_mixed_message_types(mock_get, async_client):
+    """Test chat completion with mixed message types."""
+    # Mock the response for the default model fetch
+    mock_get.return_value.status_code = 200
+    mock_get.return_value.json.return_value = {
+        "data": [{"id": "mock-model", "max_model_len": 2048}]
+    }
+
+    # Mock HTTPX response for Custom Inference API
+    mock_response = {
+        "id": "chatcmpl-test123",
+        "object": "chat.completion",
+        "model": "mock-model",
+        "choices": [
+            {
+                "index": 0,
+                "message": {
+                    "role": "assistant",
+                    "content": "This is a helpful response about the test document."
+                },
+                "finish_reason": "stop"
+            }
+        ]
+    }
+    respx.post("http://localhost:5000/v1/chat/completions").mock(return_value=httpx.Response(200, json=mock_response))
+
+    # Index some test documents
+    index_request = {
+        "index_name": "test_index",
+        "documents": [
+            {"text": "Technical documentation about software development."}
+        ]
+    }
+
+    response = await async_client.post("/index", json=index_request)
+    assert response.status_code == 200
+
+    # Test chat completion with mixed message types
+    chat_request = {
+        "index_name": "test_index",
+        "model": "mock-model",
+        "messages": [
+            {"role": "system", "content": "You are a helpful assistant."},
+            {"role": "user", "content": "I need help with coding."},
+            {"role": "assistant", "content": "I'd be happy to help with coding."},
+            {"role": "user", "content": "What about software development?"},
+            {"role": "developer", "content": "Debug: User asking about software development"}
+        ]
+    }
+
+    response = await async_client.post("/v1/chat/completions", json=chat_request)
+    assert response.status_code == 200
+    
+    response_data = response.json()
+    assert response_data["choices"][0]["message"]["content"] == "This is a helpful response about the test document."
+    assert len(response_data["source_nodes"]) > 0
+
+@pytest.mark.asyncio
+@respx.mock
+@patch("requests.get")
+async def test_chat_completions_empty_messages_list(mock_get, async_client):
+    """Test chat completion with empty messages list."""
+    # Mock the response for the default model fetch
+    mock_get.return_value.status_code = 200
+    mock_get.return_value.json.return_value = {
+        "data": [{"id": "mock-model", "max_model_len": 2048}]
+    }
+
+    # Mock HTTPX response for passthrough LLM call (in case it gets that far)
+    respx.post("http://localhost:5000/v1/chat/completions").mock(return_value=httpx.Response(400, json={"error": "Invalid request"}))
+
+    chat_request = {
+        "model": "mock-model",
+        "messages": []
+    }
+
+    response = await async_client.post("/v1/chat/completions", json=chat_request)
+    assert response.status_code == 400
+    assert "Invalid request" in response.json()["detail"]
+
+@pytest.mark.asyncio
+@respx.mock
+@patch("requests.get")
+async def test_chat_completions_with_functions(mock_get, async_client):
+    """Test chat completion with functions parameter (should passthrough)."""
+    # Mock the response for the default model fetch
+    mock_get.return_value.status_code = 200
+    mock_get.return_value.json.return_value = {
+        "data": [{"id": "mock-model", "max_model_len": 2048}]
+    }
+
+    # Mock HTTPX response for passthrough LLM call
+    mock_response = {
+        "id": "chatcmpl-functions",
+        "object": "chat.completion",
+        "created": int(time.time()),
+        "model": "mock-model",
+        "choices": [{
+            "index": 0,
+            "message": {"role": "assistant", "content": "Function-enabled response"},
+            "finish_reason": "stop"
+        }]
+    }
+    respx.post("http://localhost:5000/v1/chat/completions").mock(return_value=httpx.Response(200, json=mock_response))
+
+    # Test request with functions (should trigger passthrough)
+    chat_request = {
+        "model": "mock-model",
+        "messages": [
+            {"role": "user", "content": "Use a function to help me"}
+        ],
+        "functions": [
+            {
+                "name": "test_function",
+                "description": "A test function"
+            }
+        ]
+    }
+
+    response = await async_client.post("/v1/chat/completions", json=chat_request)
+    assert response.status_code == 200
+    
+    response_data = response.json()
+    assert response_data["choices"][0]["message"]["content"] == "Function-enabled response"
+    # Should have source_nodes field but it should be None for passthrough requests
+    assert response_data["source_nodes"] is None
\ No newline at end of file
diff --git a/presets/ragengine/tests/api/test_main.py b/presets/ragengine/tests/api/test_main.py
index 450b57a..381e685 100644
--- a/presets/ragengine/tests/api/test_main.py
+++ b/presets/ragengine/tests/api/test_main.py
@@ -22,8 +22,6 @@ from ragengine.config import DEFAULT_VECTOR_DB_PERSIST_DIR
 import os
 
 import pytest
-import pytest_asyncio
-import asyncio
 import httpx
 import respx
 import json
@@ -31,22 +29,6 @@ import re
 
 AUTO_GEN_DOC_ID_LEN = 64
 
-@pytest_asyncio.fixture
-async def async_client():
-    """Use an async HTTP client to interact with FastAPI app."""
-    async with httpx.AsyncClient(app=app, base_url="http://localhost") as client:
-        yield client
-
-@pytest.fixture(scope="session")
-def event_loop():
-    loop = asyncio.new_event_loop()
-    yield loop
-    loop.close()
-
-@pytest.fixture(autouse=True)
-def clear_index():
-    vector_store_handler.index_map.clear()
-
 @pytest.mark.asyncio
 async def test_index_documents_success(async_client):
     request_data = {
@@ -114,12 +96,6 @@ async def test_query_index_success(mock_get, async_client):
     assert response.json()["source_nodes"][0]["score"] == pytest.approx(0.5354418754577637, rel=1e-6)
     assert response.json()["source_nodes"][0]["metadata"] == {}
 
-    # Ensure HTTPX was called once
-    assert respx.calls.call_count == 1
-
-    # Ensure the model fetch was called once
-    mock_get.assert_called_once_with("http://localhost:5000/v1/models", headers=ANY)
-
     response = await async_client.get(f"/metrics")
     assert response.status_code == 200
     assert len(re.findall(r'rag_index_requests_total{status="success"} ([1-9]\d*).0', response.text)) == 1
@@ -650,4 +626,4 @@ def test_live_query_test():
 
     response = client.post("/query", json=request_data)
     assert response.status_code == 200
-"""
+"""
\ No newline at end of file
diff --git a/presets/ragengine/tests/vector_store/test_base_store.py b/presets/ragengine/tests/vector_store/test_base_store.py
index 92f3f61..1e8274d 100644
--- a/presets/ragengine/tests/vector_store/test_base_store.py
+++ b/presets/ragengine/tests/vector_store/test_base_store.py
@@ -108,6 +108,73 @@ class BaseVectorStoreTest(ABC):
         assert query_result["source_nodes"][0]["score"] == pytest.approx(self.expected_query_score, rel=1e-6)
         assert query_result["source_nodes"][0]["doc_id"] == index_doc_resp[0]
         assert respx.calls.call_count == 1  # Ensure only one LLM inference request was made
+    
+    @pytest.mark.asyncio
+    @respx.mock
+    @patch("requests.get")
+    async def test_chat_completions(self, mock_get, vector_store_manager, monkeypatch):
+        import ragengine.inference.inference
+        import ragengine.config
+        monkeypatch.setattr(ragengine.config, "LLM_INFERENCE_URL", "http://localhost:5000/v1/chat/completions")
+        monkeypatch.setattr(ragengine.inference.inference, "LLM_INFERENCE_URL", "http://localhost:5000/v1/chat/completions")
+        mock_get.return_value.status_code = 200
+        mock_get.return_value.json.return_value = {
+            "data": [{"id": "mock-model", "max_model_len": 2048}]
+        }
+
+        mock_response = {
+            "id": "chatcmpl-test123",
+            "object": "chat.completion",
+            "model": "mock-model",
+            "choices": [
+                {
+                    "index": 0,
+                    "message": {
+                        "role": "assistant",
+                        "content": "This is a helpful response about the test document."
+                    },
+                    "finish_reason": "stop"
+                }
+            ]
+        }
+        respx.post("http://localhost:5000/v1/chat/completions").mock(return_value=httpx.Response(200, json=mock_response))
+
+        documents = [
+            Document(text="First document", metadata={"type": "text"}),
+            Document(text="Second document", metadata={"type": "text"})
+        ]
+        index_doc_resp = await vector_store_manager.index_documents("test_index", documents)
+
+        chat_results = await vector_store_manager.chat_completion(
+            {
+                "index_name": "test_index",
+                "model": "mock-model",
+                "messages": [
+                    {"role": "user", "content": "What is the first document?"}
+                ],
+                "top_k": 1,
+                "temperature": 0.7,
+                "max_tokens": 100,
+            }
+        )
+
+        assert chat_results is not None
+        assert chat_results.source_nodes is not None
+        assert len(chat_results.source_nodes) == 1
+        assert chat_results.source_nodes[0].text == "First document"
+        assert chat_results.source_nodes[0].doc_id == index_doc_resp[0]
+        assert chat_results.id is not None
+        assert chat_results.model == "mock-model"
+        assert chat_results.object == "chat.completion"
+        assert chat_results.created is not None
+        assert chat_results.choices is not None
+        assert len(chat_results.choices) == 1
+        assert chat_results.choices[0].finish_reason == "stop"
+        assert chat_results.choices[0].index == 0
+        assert chat_results.choices[0].message.role == "assistant"
+        assert chat_results.choices[0].message.content == "This is a helpful response about the test document."
+
+        assert respx.calls.call_count == 1  # Ensure only one LLM inference request was made
 
     @pytest.mark.asyncio
     async def test_add_document(self, vector_store_manager):
diff --git a/test/rage2e/rag_test.go b/test/rage2e/rag_test.go
index 74e8280..acead50 100644
--- a/test/rage2e/rag_test.go
+++ b/test/rage2e/rag_test.go
@@ -103,32 +103,35 @@ var _ = Describe("RAGEngine", func() {
 		validateInferenceandRAGResource(ragengineObj.ObjectMeta, int32(numOfReplica), false)
 		validateRAGEngineCondition(ragengineObj, string(kaitov1alpha1.RAGEngineConditionTypeSucceeded), "ragengine to be ready")
 
-		indexDoc, err := createAndValidateIndexPod(ragengineObj, "index-pod")
+		indexDoc, err := createAndValidateIndexPod(ragengineObj)
 		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate IndexPod")
 		Expect(indexDoc).NotTo(BeNil(), "Index document should not be nil")
 		Expect(indexDoc["doc_id"]).NotTo(BeNil(), "Index document ID should not be nil")
 		Expect(indexDoc["text"]).NotTo(BeNil(), "Index document text should not be nil")
 		docID := indexDoc["doc_id"].(string)
 
-		searchQuerySuccess := "\\n\\nKaito is an operator that is designed to automate the AI/ML model inference or tuning workload in a Kubernetes cluster."
-		err = createAndValidateQueryPod(ragengineObj, searchQuerySuccess, true, "query-pod")
+		searchQuerySuccess := "Kaito is an operator that automates the AI/ML model inference or tuning workload in a Kubernetes cluster"
+		err = createAndValidateQueryPod(ragengineObj, searchQuerySuccess, true)
 		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate QueryPod")
 
+		err = createAndValidateQueryChatMessagesPod(ragengineObj, searchQuerySuccess, true)
+		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate QueryChatMessagesPod")
+
 		persistLogSuccess := "Successfully persisted index kaito"
-		err = createAndValidatePersistPod(ragengineObj, persistLogSuccess, "persist-pod")
+		err = createAndValidatePersistPod(ragengineObj, persistLogSuccess)
 		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate PersistPod")
 
 		loadLogSuccess := "Successfully loaded index kaito"
-		err = createAndValidateLoadPod(ragengineObj, loadLogSuccess, "load-pod")
+		err = createAndValidateLoadPod(ragengineObj, loadLogSuccess)
 		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate LoadPod")
 
-		err = createAndValidateUpdateDocumentPod(ragengineObj, docID, "update-document-pod")
+		err = createAndValidateUpdateDocumentPod(ragengineObj, docID)
 		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate UpdateDocumentPod")
 
-		err = createAndValidateDeleteDocumentPod(ragengineObj, docID, "delete-document-pod")
+		err = createAndValidateDeleteDocumentPod(ragengineObj, docID)
 		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate DeleteDocumentPod")
 
-		err = createAndValidateDeleteIndexPod(ragengineObj, "delete-index-pod")
+		err = createAndValidateDeleteIndexPod(ragengineObj)
 		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate DeleteIndexPod")
 	})
 
@@ -146,8 +149,8 @@ var _ = Describe("RAGEngine", func() {
 
 		validateWorkspaceReadiness(workspaceObj)
 
-		serviceName := workspaceObj.Name
-		serviceNamespace := workspaceObj.Namespace
+		serviceName := workspaceObj.ObjectMeta.Name
+		serviceNamespace := workspaceObj.ObjectMeta.Namespace
 		service := &v1.Service{}
 
 		_ = utils.TestingCluster.KubeClient.Get(ctx, client.ObjectKey{
@@ -157,7 +160,7 @@ var _ = Describe("RAGEngine", func() {
 
 		clusterIP := service.Spec.ClusterIP
 
-		ragengineObj := createLocalEmbeddingKaitoVLLMRAGEngine(clusterIP)
+		ragengineObj := createLocalEmbeddingKaitoVLLMRAGEngine(clusterIP, "v1/completions")
 
 		defer cleanupResources(workspaceObj, ragengineObj)
 
@@ -166,32 +169,32 @@ var _ = Describe("RAGEngine", func() {
 		validateInferenceandRAGResource(ragengineObj.ObjectMeta, int32(numOfReplica), false)
 		validateRAGEngineCondition(ragengineObj, string(kaitov1alpha1.RAGEngineConditionTypeSucceeded), "ragengine to be ready")
 
-		indexDoc, err := createAndValidateIndexPod(ragengineObj, "index-pod")
+		indexDoc, err := createAndValidateIndexPod(ragengineObj)
 		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate IndexPod")
 		Expect(indexDoc).NotTo(BeNil(), "Index document should not be nil")
 		Expect(indexDoc["doc_id"]).NotTo(BeNil(), "Index document ID should not be nil")
 		Expect(indexDoc["text"]).NotTo(BeNil(), "Index document text should not be nil")
 		docID := indexDoc["doc_id"].(string)
 
-		searchQuerySuccess := "\\nKaito is an operator that automates the AI/ML model inference or tuning workload in a Kubernetes cluster.\\n\\n\\n"
-		err = createAndValidateQueryPod(ragengineObj, searchQuerySuccess, false, "query-pod")
+		searchQuerySuccess := "Kaito is an operator that automates the AI/ML model inference or tuning workload in a Kubernetes cluster"
+		err = createAndValidateQueryPod(ragengineObj, searchQuerySuccess, false)
 		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate QueryPod")
 
 		persistLogSuccess := "Successfully persisted index kaito"
-		err = createAndValidatePersistPod(ragengineObj, persistLogSuccess, "persist-pod")
+		err = createAndValidatePersistPod(ragengineObj, persistLogSuccess)
 		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate PersistPod")
 
 		loadLogSuccess := "Successfully loaded index kaito"
-		err = createAndValidateLoadPod(ragengineObj, loadLogSuccess, "load-pod")
+		err = createAndValidateLoadPod(ragengineObj, loadLogSuccess)
 		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate LoadPod")
 
-		err = createAndValidateUpdateDocumentPod(ragengineObj, docID, "update-document-pod")
+		err = createAndValidateUpdateDocumentPod(ragengineObj, docID)
 		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate UpdateDocumentPod")
 
-		err = createAndValidateDeleteDocumentPod(ragengineObj, docID, "delete-document-pod")
+		err = createAndValidateDeleteDocumentPod(ragengineObj, docID)
 		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate DeleteDocumentPod")
 
-		err = createAndValidateDeleteIndexPod(ragengineObj, "delete-index-pod")
+		err = createAndValidateDeleteIndexPod(ragengineObj)
 		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate DeleteIndexPod")
 	})
 
@@ -233,32 +236,95 @@ var _ = Describe("RAGEngine", func() {
 		validateInferenceandRAGResource(ragengineObj.ObjectMeta, int32(numOfReplica), false)
 		validateRAGEngineCondition(ragengineObj, string(kaitov1alpha1.RAGEngineConditionTypeSucceeded), "ragengine to be ready")
 
-		indexDoc, err := createAndValidateIndexPod(ragengineObj, "index-pod")
+		indexDoc, err := createAndValidateIndexPod(ragengineObj)
 		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate IndexPod")
 		Expect(indexDoc).NotTo(BeNil(), "Index document should not be nil")
 		Expect(indexDoc["doc_id"]).NotTo(BeNil(), "Index document ID should not be nil")
 		Expect(indexDoc["text"]).NotTo(BeNil(), "Index document text should not be nil")
 		docID := indexDoc["doc_id"].(string)
 
-		searchQuerySuccess := "\\nKaito is an operator that automates the AI/ML model inference or tuning workload in a Kubernetes cluster.\\n\\n\\n"
-		err = createAndValidateQueryPod(ragengineObj, searchQuerySuccess, false, "query-pod")
+		searchQuerySuccess := "Kaito is an operator that automates the AI/ML model inference or tuning workload in a Kubernetes cluster"
+		err = createAndValidateQueryPod(ragengineObj, searchQuerySuccess, false)
 		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate QueryPod")
 
 		persistLogSuccess := "Successfully persisted index kaito"
-		err = createAndValidatePersistPod(ragengineObj, persistLogSuccess, "persist-pod")
+		err = createAndValidatePersistPod(ragengineObj, persistLogSuccess)
 		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate PersistPod")
 
 		loadLogSuccess := "Successfully loaded index kaito"
-		err = createAndValidateLoadPod(ragengineObj, loadLogSuccess, "load-pod")
+		err = createAndValidateLoadPod(ragengineObj, loadLogSuccess)
 		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate LoadPod")
 
-		err = createAndValidateUpdateDocumentPod(ragengineObj, docID, "update-document-pod")
+		err = createAndValidateUpdateDocumentPod(ragengineObj, docID)
 		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate UpdateDocumentPod")
 
-		err = createAndValidateDeleteDocumentPod(ragengineObj, docID, "delete-document-pod")
+		err = createAndValidateDeleteDocumentPod(ragengineObj, docID)
 		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate DeleteDocumentPod")
 
-		err = createAndValidateDeleteIndexPod(ragengineObj, "delete-index-pod")
+		err = createAndValidateDeleteIndexPod(ragengineObj)
+		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate DeleteIndexPod")
+	})
+
+	It("should create RAG with localembedding and kaito VLLM workspace successfully for chat completions", utils.GinkgoLabelFastCheck, func() {
+		numOfReplica := 1
+		workspaceObj := createPhi3WorkspaceWithPresetPublicModeAndVLLM(numOfReplica)
+
+		time.Sleep(30 * time.Second)
+
+		validateWorkspaceResourceStatus(workspaceObj)
+
+		validateAssociatedService(workspaceObj.ObjectMeta)
+
+		validateInferenceandRAGResource(workspaceObj.ObjectMeta, int32(numOfReplica), false)
+
+		validateWorkspaceReadiness(workspaceObj)
+
+		serviceName := workspaceObj.ObjectMeta.Name
+		serviceNamespace := workspaceObj.ObjectMeta.Namespace
+		service := &v1.Service{}
+
+		_ = utils.TestingCluster.KubeClient.Get(ctx, client.ObjectKey{
+			Namespace: serviceNamespace,
+			Name:      serviceName,
+		}, service)
+
+		clusterIP := service.Spec.ClusterIP
+
+		ragengineObj := createLocalEmbeddingKaitoVLLMRAGEngine(clusterIP, "v1/chat/completions")
+
+		defer cleanupResources(workspaceObj, ragengineObj)
+
+		validateRAGEngineCondition(ragengineObj, string(kaitov1alpha1.ConditionTypeResourceStatus), "ragengineObj resource status to be ready")
+		validateAssociatedService(ragengineObj.ObjectMeta)
+		validateInferenceandRAGResource(ragengineObj.ObjectMeta, int32(numOfReplica), false)
+		validateRAGEngineCondition(ragengineObj, string(kaitov1alpha1.RAGEngineConditionTypeSucceeded), "ragengine to be ready")
+
+		indexDoc, err := createAndValidateIndexPod(ragengineObj)
+		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate IndexPod")
+		Expect(indexDoc).NotTo(BeNil(), "Index document should not be nil")
+		Expect(indexDoc["doc_id"]).NotTo(BeNil(), "Index document ID should not be nil")
+		Expect(indexDoc["text"]).NotTo(BeNil(), "Index document text should not be nil")
+		docID := indexDoc["doc_id"].(string)
+
+		searchQuerySuccess := "Kaito is an operator that automates the AI/ML model inference or tuning workload in a Kubernetes cluster"
+		err = createAndValidateQueryChatMessagesPod(ragengineObj, searchQuerySuccess, false)
+		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate QueryChatMessagesPod")
+
+		persistLogSuccess := "Successfully persisted index kaito"
+		err = createAndValidatePersistPod(ragengineObj, persistLogSuccess)
+		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate PersistPod")
+
+		loadLogSuccess := "Successfully loaded index kaito"
+		err = createAndValidateLoadPod(ragengineObj, loadLogSuccess)
+		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate LoadPod")
+
+		err = createAndValidateUpdateDocumentPod(ragengineObj, docID)
+		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate UpdateDocumentPod")
+
+		err = createAndValidateDeleteDocumentPod(ragengineObj, docID)
+		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate DeleteDocumentPod")
+
+		err = createAndValidateDeleteIndexPod(ragengineObj)
 		Expect(err).NotTo(HaveOccurred(), "Failed to create and validate DeleteIndexPod")
 	})
 
@@ -319,7 +385,7 @@ func createAndValidateWorkspace(workspaceObj *kaitov1beta1.Workspace) {
 		Eventually(func() error {
 			return utils.TestingCluster.KubeClient.Create(ctx, workspaceObj, &client.CreateOptions{})
 		}, utils.PollTimeout, utils.PollInterval).
-			Should(Succeed(), "Failed to create workspace %s", workspaceObj.Name)
+			Should(Succeed(), "Failed to create workspace %s", workspaceObj.ObjectMeta.Name)
 	})
 }
 
@@ -328,12 +394,12 @@ func createAndValidateRAGEngine(ragEngineObj *kaitov1alpha1.RAGEngine) {
 		Eventually(func() error {
 			return utils.TestingCluster.KubeClient.Create(ctx, ragEngineObj, &client.CreateOptions{})
 		}, utils.PollTimeout, utils.PollInterval).
-			Should(Succeed(), "Failed to create ragEngine   %s", ragEngineObj.Name)
+			Should(Succeed(), "Failed to create ragEngine   %s", ragEngineObj.ObjectMeta.Name)
 
 		By("Validating ragEngine creation", func() {
 			err := utils.TestingCluster.KubeClient.Get(ctx, client.ObjectKey{
-				Namespace: ragEngineObj.Namespace,
-				Name:      ragEngineObj.Name,
+				Namespace: ragEngineObj.ObjectMeta.Namespace,
+				Name:      ragEngineObj.ObjectMeta.Name,
 			}, ragEngineObj, &client.GetOptions{})
 			Expect(err).NotTo(HaveOccurred())
 		})
@@ -387,8 +453,8 @@ func validateWorkspaceReadiness(workspaceObj *kaitov1beta1.Workspace) {
 	By("Checking the workspace status is ready", func() {
 		Eventually(func() bool {
 			err := utils.TestingCluster.KubeClient.Get(ctx, client.ObjectKey{
-				Namespace: workspaceObj.Namespace,
-				Name:      workspaceObj.Name,
+				Namespace: workspaceObj.ObjectMeta.Namespace,
+				Name:      workspaceObj.ObjectMeta.Name,
 			}, workspaceObj, &client.GetOptions{})
 
 			if err != nil {
@@ -404,9 +470,9 @@ func validateWorkspaceReadiness(workspaceObj *kaitov1beta1.Workspace) {
 	})
 }
 
-func createLocalEmbeddingKaitoVLLMRAGEngine(baseURL string) *kaitov1alpha1.RAGEngine {
+func createLocalEmbeddingKaitoVLLMRAGEngine(baseURL, llmPath string) *kaitov1alpha1.RAGEngine {
 	ragEngineObj := &kaitov1alpha1.RAGEngine{}
-	serviceURL := fmt.Sprintf("http://%s/v1/completions", baseURL)
+	serviceURL := fmt.Sprintf("http://%s/%s", baseURL, llmPath)
 	By("Creating RAG with localembedding and kaito vllm inference", func() {
 		uniqueID := fmt.Sprint("rag-", rand.Intn(1000))
 		ragEngineObj = GenerateLocalEmbeddingRAGEngineManifest(uniqueID, namespaceName, "Standard_NV36ads_A10_v5", "BAAI/bge-small-en-v1.5",
@@ -444,7 +510,7 @@ func createLocalPreferredNodesRAGEngine(baseURL, preferredNode string) *kaitov1a
 
 func createLocalEmbeddingHFURLRAGEngine() *kaitov1alpha1.RAGEngine {
 	ragEngineObj := &kaitov1alpha1.RAGEngine{}
-	hfURL := "https://api-inference.huggingface.co/models/HuggingFaceH4/zephyr-7b-beta/v1/completions"
+	hfURL := "https://router.huggingface.co/featherless-ai/v1/chat/completions"
 	By("Creating RAG with localembedding and huggingface API", func() {
 		uniqueID := fmt.Sprint("rag-", rand.Intn(1000))
 		ragEngineObj = GenerateLocalEmbeddingRAGEngineManifest(uniqueID, namespaceName, "Standard_NV36ads_A10_v5", "BAAI/bge-small-en-v1.5",
@@ -478,9 +544,9 @@ func cleanupResources(
 		} else {
 			if ragengineObj != nil {
 				GinkgoWriter.Printf("Test failed, keep Workspace %s and RAGEngine %s\n",
-					workspaceObj.Name, ragengineObj.Name)
+					workspaceObj.ObjectMeta.Name, ragengineObj.ObjectMeta.Name)
 			} else {
-				GinkgoWriter.Printf("Test failed, keep Workspace %s\n", workspaceObj.Name)
+				GinkgoWriter.Printf("Test failed, keep Workspace %s\n", workspaceObj.ObjectMeta.Name)
 			}
 		}
 	})
@@ -491,8 +557,8 @@ func validateWorkspaceResourceStatus(workspaceObj *kaitov1beta1.Workspace) {
 	By("Checking the resource status", func() {
 		Eventually(func() bool {
 			err := utils.TestingCluster.KubeClient.Get(ctx, client.ObjectKey{
-				Namespace: workspaceObj.Namespace,
-				Name:      workspaceObj.Name,
+				Namespace: workspaceObj.ObjectMeta.Namespace,
+				Name:      workspaceObj.ObjectMeta.Name,
 			}, workspaceObj, &client.GetOptions{})
 
 			if err != nil {
@@ -513,21 +579,21 @@ func deleteRAGEngine(ragengineObj *kaitov1alpha1.RAGEngine) error {
 		Eventually(func() error {
 			// Check if the workspace exists
 			err := utils.TestingCluster.KubeClient.Get(ctx, client.ObjectKey{
-				Namespace: ragengineObj.Namespace,
-				Name:      ragengineObj.Name,
+				Namespace: ragengineObj.ObjectMeta.Namespace,
+				Name:      ragengineObj.ObjectMeta.Name,
 			}, ragengineObj)
 
 			if errors.IsNotFound(err) {
-				GinkgoWriter.Printf("RAGEngine %s does not exist, no need to delete\n", ragengineObj.Name)
+				GinkgoWriter.Printf("RAGEngine %s does not exist, no need to delete\n", ragengineObj.ObjectMeta.Name)
 				return nil
 			}
 			if err != nil {
-				return fmt.Errorf("error checking if ragengine %s exists: %v", ragengineObj.Name, err)
+				return fmt.Errorf("error checking if ragengine %s exists: %v", ragengineObj.ObjectMeta.Name, err)
 			}
 
 			err = utils.TestingCluster.KubeClient.Delete(ctx, ragengineObj, &client.DeleteOptions{})
 			if err != nil {
-				return fmt.Errorf("failed to delete ragengine %s: %v", ragengineObj.Name, err)
+				return fmt.Errorf("failed to delete ragengine %s: %v", ragengineObj.ObjectMeta.Name, err)
 			}
 			return nil
 		}, utils.PollTimeout, utils.PollInterval).Should(Succeed(), "Failed to delete ragengine")
@@ -617,8 +683,8 @@ func validateRAGEngineCondition(ragengineObj *kaitov1alpha1.RAGEngine, condition
 	By(fmt.Sprintf("Checking %s", description), func() {
 		Eventually(func() bool {
 			err := utils.TestingCluster.KubeClient.Get(ctx, client.ObjectKey{
-				Namespace: ragengineObj.Namespace,
-				Name:      ragengineObj.Name,
+				Namespace: ragengineObj.ObjectMeta.Namespace,
+				Name:      ragengineObj.ObjectMeta.Name,
 			}, ragengineObj, &client.GetOptions{})
 			if err != nil {
 				return false
@@ -637,21 +703,21 @@ func deleteWorkspace(workspaceObj *kaitov1beta1.Workspace) error {
 		Eventually(func() error {
 			// Check if the workspace exists
 			err := utils.TestingCluster.KubeClient.Get(ctx, client.ObjectKey{
-				Namespace: workspaceObj.Namespace,
-				Name:      workspaceObj.Name,
+				Namespace: workspaceObj.ObjectMeta.Namespace,
+				Name:      workspaceObj.ObjectMeta.Name,
 			}, workspaceObj)
 
 			if errors.IsNotFound(err) {
-				GinkgoWriter.Printf("Workspace %s does not exist, no need to delete\n", workspaceObj.Name)
+				GinkgoWriter.Printf("Workspace %s does not exist, no need to delete\n", workspaceObj.ObjectMeta.Name)
 				return nil
 			}
 			if err != nil {
-				return fmt.Errorf("error checking if workspace %s exists: %v", workspaceObj.Name, err)
+				return fmt.Errorf("error checking if workspace %s exists: %v", workspaceObj.ObjectMeta.Name, err)
 			}
 
 			err = utils.TestingCluster.KubeClient.Delete(ctx, workspaceObj, &client.DeleteOptions{})
 			if err != nil {
-				return fmt.Errorf("failed to delete workspace %s: %v", workspaceObj.Name, err)
+				return fmt.Errorf("failed to delete workspace %s: %v", workspaceObj.ObjectMeta.Name, err)
 			}
 			return nil
 		}, utils.PollTimeout, utils.PollInterval).Should(Succeed(), "Failed to delete workspace")
@@ -660,10 +726,8 @@ func deleteWorkspace(workspaceObj *kaitov1beta1.Workspace) error {
 	return nil
 }
 
-func createAndValidateIndexPod(ragengineObj *kaitov1alpha1.RAGEngine, podName string) (map[string]any, error) {
-	indexResp := []map[string]any{}
-	By("Creating index pod", func() {
-		curlCommand := `curl -X POST ` + ragengineObj.Name + `:80/index \
+func createAndValidateIndexPod(ragengineObj *kaitov1alpha1.RAGEngine) (map[string]any, error) {
+	curlCommand := `curl -X POST ` + ragengineObj.ObjectMeta.Name + `:80/index \
 -H "Content-Type: application/json" \
 -d '{
     "index_name": "kaito",
@@ -674,60 +738,21 @@ func createAndValidateIndexPod(ragengineObj *kaitov1alpha1.RAGEngine, podName st
         }
     ]
 }'`
-		pod := GenerateCURLPodManifest(podName, curlCommand, ragengineObj.Namespace)
-		Eventually(func() error {
-			return utils.TestingCluster.KubeClient.Create(ctx, pod, &client.CreateOptions{})
-		}, utils.PollTimeout, utils.PollInterval).
-			Should(Succeed(), "Failed to create index pod")
-	})
-
-	By("Checking the index logs", func() {
-		Eventually(func() bool {
-			coreClient, err := utils.GetK8sClientset()
-			if err != nil {
-				GinkgoWriter.Printf("Failed to create core client: %v\n", err)
-				return false
-			}
-
-			logs, err := utils.GetPodLogs(coreClient, ragengineObj.Namespace, podName, "")
-			if err != nil {
-				GinkgoWriter.Printf("Failed to get logs from pod %s: %v\n", podName, err)
-				return false
-			}
-
-			GinkgoWriter.Printf("Index pod logs: %s\n", logs)
-
-			startIndex := strings.Index(logs, "[")
-			endIndex := strings.LastIndex(logs, "]")
-			if startIndex == -1 || endIndex == -1 || startIndex >= endIndex {
-				GinkgoWriter.Printf("Invalid index logs format in pod %s: %s\n", podName, logs)
-				return false
-			}
-
-			apiResp := logs[startIndex : endIndex+1]
-			GinkgoWriter.Printf("Parsed API response: %s\n", apiResp)
-
-			err = json.Unmarshal([]byte(apiResp), &indexResp)
-			if err != nil {
-				GinkgoWriter.Printf("Failed to unmarshal pod logs to index response %s: %v\n", podName, err)
-				return false
-			}
-
-			if len(indexResp) == 0 {
-				GinkgoWriter.Printf("No index logs found in pod %s\n", podName)
-				return false
-			}
-
-			return strings.Contains(logs, "Kaito is an operator that automates the AI/ML model inference or tuning workload in a Kubernetes cluster")
-		}, 4*time.Minute, utils.PollInterval).Should(BeTrue(), "Failed to wait for index logs to be ready")
-	})
-
-	return indexResp[0], nil
+	opts := PodValidationOptions{
+		PodName:            fmt.Sprintf("index-pod-%s", utils.GenerateRandomString()),
+		CurlCommand:        curlCommand,
+		Namespace:          ragengineObj.ObjectMeta.Namespace,
+		ExpectedLogContent: "Kaito is an operator that automates the AI/ML model inference or tuning workload in a Kubernetes cluster",
+		WaitForRunning:     false,
+		ParseJSONResponse:  true,
+		JSONStartMarker:    "[",
+		JSONEndMarker:      "]",
+	}
+	return createAndValidateAPIPod(ragengineObj, opts)
 }
 
-func createAndValidateUpdateDocumentPod(ragengineObj *kaitov1alpha1.RAGEngine, docID string, podName string) error {
-	By("Creating update document pod", func() {
-		curlCommand := `curl -X POST ` + ragengineObj.Name + `:80/indexes/kaito/documents \
+func createAndValidateUpdateDocumentPod(ragengineObj *kaitov1alpha1.RAGEngine, docID string) error {
+	curlCommand := `curl -X POST ` + ragengineObj.ObjectMeta.Name + `:80/indexes/kaito/documents \
 -H "Content-Type: application/json" \
 -d '{
     "documents": [
@@ -738,122 +763,66 @@ func createAndValidateUpdateDocumentPod(ragengineObj *kaitov1alpha1.RAGEngine, d
         }
     ]
 }'`
-		pod := GenerateCURLPodManifest(podName, curlCommand, ragengineObj.Namespace)
-		Eventually(func() error {
-			return utils.TestingCluster.KubeClient.Create(ctx, pod, &client.CreateOptions{})
-		}, utils.PollTimeout, utils.PollInterval).
-			Should(Succeed(), "Failed to create index pod")
-	})
-
-	By("Checking the index logs", func() {
-		Eventually(func() bool {
-			coreClient, err := utils.GetK8sClientset()
-			if err != nil {
-				GinkgoWriter.Printf("Failed to create core client: %v\n", err)
-				return false
-			}
-
-			logs, err := utils.GetPodLogs(coreClient, ragengineObj.Namespace, podName, "")
-			if err != nil {
-				GinkgoWriter.Printf("Failed to get logs from pod %s: %v\n", podName, err)
-				return false
-			}
-
-			GinkgoWriter.Printf("Index pod logs: %s\n", logs)
-
-			return strings.Contains(logs, `"updated_documents":[{"doc_id":"`+docID+`"`)
-		}, 4*time.Minute, utils.PollInterval).Should(BeTrue(), "Failed to wait for update document logs to be ready")
-	})
-
-	return nil
+	opts := PodValidationOptions{
+		PodName:            fmt.Sprintf("update-document-pod-%s", utils.GenerateRandomString()),
+		CurlCommand:        curlCommand,
+		Namespace:          ragengineObj.ObjectMeta.Namespace,
+		ExpectedLogContent: `"updated_documents":[{"doc_id":"` + docID + `"`,
+		WaitForRunning:     false,
+		ParseJSONResponse:  false,
+	}
+	_, err := createAndValidateAPIPod(ragengineObj, opts)
+	return err
 }
 
-func createAndValidateDeleteDocumentPod(ragengineObj *kaitov1alpha1.RAGEngine, docID string, podName string) error {
-	By("Creating delete document pod", func() {
-		curlCommand := `curl -X POST ` + ragengineObj.Name + `:80/indexes/kaito/documents/delete \
+func createAndValidateDeleteDocumentPod(ragengineObj *kaitov1alpha1.RAGEngine, docID string) error {
+	curlCommand := `curl -X POST ` + ragengineObj.ObjectMeta.Name + `:80/indexes/kaito/documents/delete \
 -H "Content-Type: application/json" \
 -d '{"doc_ids": ["` + docID + `"]}'`
-		pod := GenerateCURLPodManifest(podName, curlCommand, ragengineObj.Namespace)
-		Eventually(func() error {
-			return utils.TestingCluster.KubeClient.Create(ctx, pod, &client.CreateOptions{})
-		}, utils.PollTimeout, utils.PollInterval).
-			Should(Succeed(), "Failed to create index pod")
-	})
-
-	By("Checking the index logs", func() {
-		Eventually(func() bool {
-			coreClient, err := utils.GetK8sClientset()
-			if err != nil {
-				GinkgoWriter.Printf("Failed to create core client: %v\n", err)
-				return false
-			}
-
-			logs, err := utils.GetPodLogs(coreClient, ragengineObj.Namespace, podName, "")
-			if err != nil {
-				GinkgoWriter.Printf("Failed to get logs from pod %s: %v\n", podName, err)
-				return false
-			}
-
-			GinkgoWriter.Printf("Index pod logs: %s\n", logs)
-
-			return strings.Contains(logs, `"deleted_doc_ids":["`+docID+`"]`)
-		}, 4*time.Minute, utils.PollInterval).Should(BeTrue(), "Failed to wait for delete document logs to be ready")
-	})
-
-	return nil
+	opts := PodValidationOptions{
+		PodName:            fmt.Sprintf("delete-document-pod-%s", utils.GenerateRandomString()),
+		CurlCommand:        curlCommand,
+		Namespace:          ragengineObj.ObjectMeta.Namespace,
+		ExpectedLogContent: `"deleted_doc_ids":["` + docID + `"]`,
+		WaitForRunning:     false,
+		ParseJSONResponse:  false,
+	}
+	_, err := createAndValidateAPIPod(ragengineObj, opts)
+	return err
 }
 
-func createAndValidateDeleteIndexPod(ragengineObj *kaitov1alpha1.RAGEngine, podName string) error {
-	By("Creating delete index pod", func() {
-		curlCommand := `curl -X DELETE ` + ragengineObj.Name + `:80/indexes/kaito \
+func createAndValidateDeleteIndexPod(ragengineObj *kaitov1alpha1.RAGEngine) error {
+	curlCommand := `curl -X DELETE ` + ragengineObj.ObjectMeta.Name + `:80/indexes/kaito \
 -H "Content-Type: application/json"`
-		pod := GenerateCURLPodManifest(podName, curlCommand, ragengineObj.Namespace)
-		Eventually(func() error {
-			return utils.TestingCluster.KubeClient.Create(ctx, pod, &client.CreateOptions{})
-		}, utils.PollTimeout, utils.PollInterval).
-			Should(Succeed(), "Failed to create delete index pod")
-	})
-
-	By("Checking the delete index logs", func() {
-		Eventually(func() bool {
-			coreClient, err := utils.GetK8sClientset()
-			if err != nil {
-				GinkgoWriter.Printf("Failed to create core client: %v\n", err)
-				return false
-			}
-
-			logs, err := utils.GetPodLogs(coreClient, ragengineObj.Namespace, podName, "")
-			if err != nil {
-				GinkgoWriter.Printf("Failed to get logs from pod %s: %v\n", podName, err)
-				return false
-			}
-
-			GinkgoWriter.Printf("Index pod logs: %s\n", logs)
-
-			return strings.Contains(logs, "Successfully deleted index kaito")
-		}, 4*time.Minute, utils.PollInterval).Should(BeTrue(), "Failed to wait for index logs to be ready")
-	})
-
-	return nil
+	opts := PodValidationOptions{
+		PodName:            fmt.Sprintf("delete-index-pod-%s", utils.GenerateRandomString()),
+		CurlCommand:        curlCommand,
+		Namespace:          ragengineObj.ObjectMeta.Namespace,
+		ExpectedLogContent: "Successfully deleted index kaito",
+		WaitForRunning:     false,
+		ParseJSONResponse:  false,
+	}
+	_, err := createAndValidateAPIPod(ragengineObj, opts)
+	return err
 }
 
-func createAndValidateQueryPod(ragengineObj *kaitov1alpha1.RAGEngine, expectedSearchQueries string, remote bool, podName string) error {
-	By("Creating query pod", func() {
-		var curlCommand string
-		// Note: Request without model specified should still succeed with vLLM. As model name is dynamically fetched.
-		if remote {
-			curlCommand = `curl -X POST ` + ragengineObj.Name + `:80/query \
+func createAndValidateQueryPod(ragengineObj *kaitov1alpha1.RAGEngine, expectedSearchQueries string, remote bool) error {
+	var curlCommand string
+	// Note: Request without model specified should still succeed with vLLM. As model name is dynamically fetched.
+	if remote {
+		curlCommand = `curl -X POST ` + ragengineObj.ObjectMeta.Name + `:80/query \
 -H "Content-Type: application/json" \
 -d '{
 	"index_name": "kaito",
+	"model": "HuggingFaceH4/zephyr-7b-beta",
     "query": "what is kaito?",
     "llm_params": {
       "max_tokens": 50,
       "temperature": 0
     }
 }'`
-		} else {
-			curlCommand = `curl -X POST ` + ragengineObj.Name + `:80/query \
+	} else {
+		curlCommand = `curl -X POST ` + ragengineObj.ObjectMeta.Name + `:80/query \
 -H "Content-Type: application/json" \
 -d '{
 	"index_name": "kaito",
@@ -864,129 +833,195 @@ func createAndValidateQueryPod(ragengineObj *kaitov1alpha1.RAGEngine, expectedSe
       "temperature": 0
     }
 }'`
+	}
+	opts := PodValidationOptions{
+		PodName:            fmt.Sprintf("query-pod-%s", utils.GenerateRandomString()),
+		CurlCommand:        curlCommand,
+		Namespace:          ragengineObj.ObjectMeta.Namespace,
+		ExpectedLogContent: expectedSearchQueries,
+		WaitForRunning:     true,
+		ParseJSONResponse:  false,
+	}
+	_, err := createAndValidateAPIPod(ragengineObj, opts)
+	return err
+}
+
+func createAndValidateQueryChatMessagesPod(ragengineObj *kaitov1alpha1.RAGEngine, expectedSearchQueries string, remote bool) error {
+	var curlCommand string
+	// Note: Request without model specified should still succeed with vLLM. As model name is dynamically fetched.
+	if remote {
+		curlCommand = `curl -X POST ` + ragengineObj.ObjectMeta.Name + `:80/v1/chat/completions \
+-H "Content-Type: application/json" \
+-d '{
+	"index_name": "kaito",
+	"model": "HuggingFaceH4/zephyr-7b-beta",
+    "messages": [
+		{
+			"role": "user",
+			"content": "what is kaito?"
 		}
-		pod := GenerateCURLPodManifest(podName, curlCommand, ragengineObj.Namespace)
-		Eventually(func() error {
-			return utils.TestingCluster.KubeClient.Create(ctx, pod, &client.CreateOptions{})
-		}, utils.PollTimeout, utils.PollInterval).
-			Should(Succeed(), "Failed to create query pod")
-	})
+	],
+    "max_tokens": 50,
+    "temperature": 0
+}'`
+	} else {
+		curlCommand = `curl -X POST ` + ragengineObj.ObjectMeta.Name + `:80/v1/chat/completions \
+-H "Content-Type: application/json" \
+-d '{
+	"index_name": "kaito",
+    "model": "phi-3-mini-128k-instruct",
+    "messages": [
+		{
+			"role": "user",
+			"content": "what is kaito?"
+		}
+	],
+    "max_tokens": 50,
+    "temperature": 0
+}'`
+	}
+	opts := PodValidationOptions{
+		PodName:            fmt.Sprintf("chat-completions-pod-%s", utils.GenerateRandomString()),
+		CurlCommand:        curlCommand,
+		Namespace:          ragengineObj.ObjectMeta.Namespace,
+		ExpectedLogContent: expectedSearchQueries,
+		WaitForRunning:     true,
+		ParseJSONResponse:  false,
+	}
+	_, err := createAndValidateAPIPod(ragengineObj, opts)
+	return err
+}
 
-	By("Waiting for query pod to be running", func() {
-		Eventually(func() bool {
-			pod := &v1.Pod{}
-			err := utils.TestingCluster.KubeClient.Get(ctx, client.ObjectKey{
-				Namespace: ragengineObj.Namespace,
-				Name:      podName,
-			}, pod)
-			if err != nil {
-				return false
-			}
-			return pod.Status.Phase == v1.PodRunning || pod.Status.Phase == v1.PodSucceeded
-		}, 5*time.Minute, utils.PollInterval).Should(BeTrue(), "Query pod did not reach Running or Succeeded state")
-	})
+func createAndValidatePersistPod(ragengineObj *kaitov1alpha1.RAGEngine, expectedPersistResult string) error {
+	curlCommand := `curl -X POST ` + ragengineObj.ObjectMeta.Name + `:80/persist/kaito`
+	opts := PodValidationOptions{
+		PodName:            fmt.Sprintf("persist-pod-%s", utils.GenerateRandomString()),
+		CurlCommand:        curlCommand,
+		Namespace:          ragengineObj.ObjectMeta.Namespace,
+		ExpectedLogContent: expectedPersistResult,
+		WaitForRunning:     true,
+		ParseJSONResponse:  false,
+	}
+	_, err := createAndValidateAPIPod(ragengineObj, opts)
+	return err
+}
 
-	By("Checking the query logs", func() {
-		Eventually(func() bool {
-			coreClient, err := utils.GetK8sClientset()
-			if err != nil {
-				GinkgoWriter.Printf("Failed to create core client: %v\n", err)
-				return false
-			}
-			logs, err := utils.GetPodLogs(coreClient, ragengineObj.Namespace, podName, "")
-			if err != nil {
-				GinkgoWriter.Printf("Failed to get logs from pod %s: %v\n", podName, err)
-				return false
-			}
-			return strings.Contains(logs, expectedSearchQueries)
-		}, 4*time.Minute, utils.PollInterval).Should(BeTrue(), "Failed to wait for query logs to be ready")
-	})
+func createAndValidateLoadPod(ragengineObj *kaitov1alpha1.RAGEngine, expectedLoadResult string) error {
+	curlCommand := `curl -X POST ` + ragengineObj.ObjectMeta.Name + `:80/load/kaito?overwrite=True`
+	opts := PodValidationOptions{
+		PodName:            fmt.Sprintf("load-pod-%s", utils.GenerateRandomString()),
+		CurlCommand:        curlCommand,
+		Namespace:          ragengineObj.ObjectMeta.Namespace,
+		ExpectedLogContent: expectedLoadResult,
+		WaitForRunning:     true,
+		ParseJSONResponse:  false,
+	}
+	_, err := createAndValidateAPIPod(ragengineObj, opts)
+	return err
+}
 
-	return nil
+// PodValidationOptions holds configuration for pod validation
+type PodValidationOptions struct {
+	PodName            string
+	CurlCommand        string
+	Namespace          string
+	ExpectedLogContent string
+	WaitForRunning     bool
+	ParseJSONResponse  bool
+	JSONStartMarker    string
+	JSONEndMarker      string
 }
 
-func createAndValidatePersistPod(ragengineObj *kaitov1alpha1.RAGEngine, expectedPersistResult string, podName string) error {
-	By("Creating Persist pod", func() {
-		curlCommand := `curl -X POST ` + ragengineObj.Name + `:80/persist/kaito`
-		pod := GenerateCURLPodManifest(podName, curlCommand, ragengineObj.Namespace)
-		Eventually(func() error {
-			return utils.TestingCluster.KubeClient.Create(ctx, pod, &client.CreateOptions{})
-		}, utils.PollTimeout, utils.PollInterval).
-			Should(Succeed(), "Failed to create persist pod")
-	})
+// createAndValidateAPIPod is a generic function to create and validate API test pods
+func createAndValidateAPIPod(ragengineObj *kaitov1alpha1.RAGEngine, opts PodValidationOptions) (map[string]any, error) {
+	var jsonResp []map[string]any
 
-	By("Waiting for persist pod to be running", func() {
-		Eventually(func() bool {
-			pod := &v1.Pod{}
-			err := utils.TestingCluster.KubeClient.Get(ctx, client.ObjectKey{
-				Namespace: ragengineObj.Namespace,
-				Name:      podName,
-			}, pod)
+	By(fmt.Sprintf("Creating %s", opts.PodName), func() {
+		pod := GenerateCURLPodManifest(opts.PodName, opts.CurlCommand, opts.Namespace)
+		Eventually(func() error {
+			err := utils.TestingCluster.KubeClient.Create(ctx, pod, &client.CreateOptions{})
 			if err != nil {
-				return false
+				GinkgoWriter.Printf("Failed to create pod %s: %v\n", opts.PodName, err)
+				return err
 			}
-			return pod.Status.Phase == v1.PodRunning || pod.Status.Phase == v1.PodSucceeded
-		}, 5*time.Minute, utils.PollInterval).Should(BeTrue(), "Persist pod did not reach Running or Succeeded state")
+			return nil
+		}, utils.PollTimeout, utils.PollInterval).
+			Should(Succeed(), fmt.Sprintf("Failed to create %s", opts.PodName))
 	})
 
-	By("Checking the persist logs", func() {
+	if opts.WaitForRunning {
+		By(fmt.Sprintf("Waiting for %s to be running", opts.PodName), func() {
+			Eventually(func() bool {
+				pod := &v1.Pod{}
+				err := utils.TestingCluster.KubeClient.Get(ctx, client.ObjectKey{
+					Namespace: opts.Namespace,
+					Name:      opts.PodName,
+				}, pod)
+				if err != nil {
+					GinkgoWriter.Printf("Failed to get pod %s: %v\n", opts.PodName, err)
+					return false
+				}
+				return pod.Status.Phase == v1.PodRunning || pod.Status.Phase == v1.PodSucceeded
+			}, 5*time.Minute, utils.PollInterval).Should(BeTrue(), fmt.Sprintf("%s did not reach Running or Succeeded state", opts.PodName))
+		})
+	}
+
+	By(fmt.Sprintf("Checking the %s logs", opts.PodName), func() {
 		Eventually(func() bool {
 			coreClient, err := utils.GetK8sClientset()
 			if err != nil {
 				GinkgoWriter.Printf("Failed to create core client: %v\n", err)
 				return false
 			}
-			logs, err := utils.GetPodLogs(coreClient, ragengineObj.Namespace, podName, "")
+
+			logs, err := utils.GetPodLogs(coreClient, opts.Namespace, opts.PodName, "")
 			if err != nil {
-				GinkgoWriter.Printf("Failed to get logs from pod %s: %v\n", podName, err)
+				GinkgoWriter.Printf("Failed to get logs from pod %s: %v\n", opts.PodName, err)
 				return false
 			}
-			return strings.Contains(logs, expectedPersistResult)
-		}, 4*time.Minute, utils.PollInterval).Should(BeTrue(), "Failed to wait for persist logs to be ready")
-	})
 
-	return nil
-}
+			GinkgoWriter.Printf("%s logs: %s\n", opts.PodName, logs)
 
-func createAndValidateLoadPod(ragengineObj *kaitov1alpha1.RAGEngine, expectedLoadResult string, podName string) error {
-	By("Creating Load Pod", func() {
-		curlCommand := `curl -X POST ` + ragengineObj.Name + `:80/load/kaito?overwrite=True`
-		pod := GenerateCURLPodManifest(podName, curlCommand, ragengineObj.Namespace)
-		Eventually(func() error {
-			return utils.TestingCluster.KubeClient.Create(ctx, pod, &client.CreateOptions{})
-		}, utils.PollTimeout, utils.PollInterval).
-			Should(Succeed(), "Failed to create load pod")
-	})
-	// Wait for the pod to be running before attempting to fetch logs.
-	By("Waiting for load pod to be running", func() {
-		Eventually(func() bool {
-			pod := &v1.Pod{}
-			err := utils.TestingCluster.KubeClient.Get(ctx, client.ObjectKey{
-				Namespace: ragengineObj.Namespace,
-				Name:      podName,
-			}, pod)
-			if err != nil {
-				return false
-			}
-			return pod.Status.Phase == v1.PodRunning || pod.Status.Phase == v1.PodSucceeded
-		}, 5*time.Minute, utils.PollInterval).Should(BeTrue(), "Load pod did not reach Running or Succeeded state")
-	})
-	By("Checking the load logs", func() {
-		Eventually(func() bool {
-			coreClient, err := utils.GetK8sClientset()
-			if err != nil {
-				GinkgoWriter.Printf("Failed to create core client: %v\n", err)
-				return false
-			}
-			logs, err := utils.GetPodLogs(coreClient, ragengineObj.Namespace, podName, "")
-			if err != nil {
-				GinkgoWriter.Printf("Failed to get logs from pod %s: %v\n", podName, err)
-				return false
+			if opts.ParseJSONResponse {
+				startMarker := opts.JSONStartMarker
+				endMarker := opts.JSONEndMarker
+				if startMarker == "" {
+					startMarker = "["
+				}
+				if endMarker == "" {
+					endMarker = "]"
+				}
+
+				startIndex := strings.Index(logs, startMarker)
+				endIndex := strings.LastIndex(logs, endMarker)
+				if startIndex == -1 || endIndex == -1 || startIndex >= endIndex {
+					GinkgoWriter.Printf("Invalid JSON format in pod %s: %s\n", opts.PodName, logs)
+					return false
+				}
+
+				apiResp := logs[startIndex : endIndex+1]
+				GinkgoWriter.Printf("Parsed API response: %s\n", apiResp)
+
+				err = json.Unmarshal([]byte(apiResp), &jsonResp)
+				if err != nil {
+					GinkgoWriter.Printf("Failed to unmarshal pod logs to JSON response %s: %v\n", opts.PodName, err)
+					return false
+				}
+
+				if len(jsonResp) == 0 {
+					GinkgoWriter.Printf("No JSON response found in pod %s\n", opts.PodName)
+					return false
+				}
 			}
-			return strings.Contains(logs, expectedLoadResult)
-		}, 4*time.Minute, utils.PollInterval).Should(BeTrue(), "Failed to wait for load logs to be ready")
+
+			return strings.Contains(logs, opts.ExpectedLogContent)
+		}, 4*time.Minute, utils.PollInterval).Should(BeTrue(), fmt.Sprintf("Failed to wait for %s logs to be ready", opts.PodName))
 	})
-	return nil
+
+	if opts.ParseJSONResponse && len(jsonResp) > 0 {
+		return jsonResp[0], nil
+	}
+	return nil, nil
 }
 
 func GenerateCURLPodManifest(podName, curlCommand, namespace string) *v1.Pod {