Deploy files from GitHub repository

app/rag/__init__.py

@@ -49,11 +49,11 @@ inferencer_config = InferencerConfig(
 )
 
 document_retriever = LangChainRetriever(
-        embedding_model="sentence-transformers/all-MiniLM-L6-v2",
+        embedding_model="BAAI/bge-large-en",
         vectorstore_type="chroma",
         vectorstore_path="vectorstore/",
         use_hybrid_search=True,
-        chunk_size=1000,
+        chunk_size=3000,
         chunk_overlap=200
 )
 

app/rag/agents/customer_service_agent.py

@@ -8,9 +8,9 @@ class CSAgent(Agent):
         self.prompt_template = prompt_template
         self.file_paths = [
             "../documents/bpjs.pdf",
-            "../documents/pph21.pdf",
-            "../documents/lembur.pdf",
-            "../documents/uu13.pdf",
+            # "../documents/pph21.pdf",
+            # "../documents/lembur.pdf",
+            # "../documents/uu13.pdf",
             "../documents/file.pdf",
         ]
     async def load_documents(self):

app/rag/retriever/langchain_retriever.py

@@ -175,6 +175,7 @@ class LangChainRetriever(BaseRetriever):
                 vectorstore=self.vectorstore,
                 search_kwargs={"k": 10}
             )
+
             self.retriever = EnsembleRetriever(
                 retrievers=[vector_retriever, self.bm25_retriever],
                 weights=[0.5, 0.5]  # Equal weight to both retrievers
@@ -197,6 +198,7 @@ class LangChainRetriever(BaseRetriever):
                 None, self.retriever.get_relevant_documents, query
             )
             retrieved_docs = retrieved_docs[:k]
+
             scores = [0.9 - (i * 0.1) for i in range(len(retrieved_docs))]
 
             retrieval_time = time.time() - start_time

space/app/rag/__init__.py

@@ -49,11 +49,11 @@ inferencer_config = InferencerConfig(
 )
 
 document_retriever = LangChainRetriever(
-        embedding_model="sentence-transformers/all-MiniLM-L6-v2",
+        embedding_model="BAAI/bge-large-en",
         vectorstore_type="chroma",
         vectorstore_path="vectorstore/",
         use_hybrid_search=True,
-        chunk_size=1000,
+        chunk_size=3000,
         chunk_overlap=200
 )
 

space/app/rag/agents/customer_service_agent.py

@@ -8,9 +8,9 @@ class CSAgent(Agent):
         self.prompt_template = prompt_template
         self.file_paths = [
             "../documents/bpjs.pdf",
-            "../documents/pph21.pdf",
-            "../documents/lembur.pdf",
-            "../documents/uu13.pdf",
+            # "../documents/pph21.pdf",
+            # "../documents/lembur.pdf",
+            # "../documents/uu13.pdf",
             "../documents/file.pdf",
         ]
     async def load_documents(self):

space/app/rag/retriever/langchain_retriever.py

@@ -175,6 +175,7 @@ class LangChainRetriever(BaseRetriever):
                 vectorstore=self.vectorstore,
                 search_kwargs={"k": 10}
             )
+
             self.retriever = EnsembleRetriever(
                 retrievers=[vector_retriever, self.bm25_retriever],
                 weights=[0.5, 0.5]  # Equal weight to both retrievers
@@ -197,6 +198,7 @@ class LangChainRetriever(BaseRetriever):
                 None, self.retriever.get_relevant_documents, query
             )
             retrieved_docs = retrieved_docs[:k]
+
             scores = [0.9 - (i * 0.1) for i in range(len(retrieved_docs))]
 
             retrieval_time = time.time() - start_time

space/space/app/rag/agents/agents.py

@@ -0,0 +1,16 @@
+from rag.pipeline.language_model import LM
+from rag.inference.inferencer import Inferencer
+from abc import ABC, abstractmethod
+class Agent(ABC):
+    def __init__(self, inferencer:Inferencer, prompt_template = [
+        {
+            "role" : "system",
+            "content":"You are an agent that doing some specic task"
+        }
+    ]):
+        self.inferencer = inferencer
+        self.inferencer.model.prompt_template = prompt_template
+        self.prompt = prompt_template
+    @abstractmethod
+    async def get_result(self):
+        pass

space/space/app/rag/agents/customer_service_agent.py

@@ -0,0 +1,33 @@
+from rag.agents.agents import Agent
+from rag.inference.inferencer import Inferencer
+
+class CSAgent(Agent):
+    def __init__(self, inferencer : Inferencer , prompt_template):
+        super().__init__(inferencer, prompt_template)
+        self.inferencer = inferencer
+        self.prompt_template = prompt_template
+        self.file_paths = [
+            "../documents/bpjs.pdf",
+            # "../documents/pph21.pdf",
+            # "../documents/lembur.pdf",
+            # "../documents/uu13.pdf",
+            "../documents/file.pdf",
+        ]
+    async def load_documents(self):
+        for file_path in self.file_paths:
+            await self.add_doc(file_path)
+        
+    async def add_doc(self, file_path):
+        result = await self.inferencer.retriever.add_document_from_file(file_path)
+        if result.success:
+                print(f"Successfully processed: {result.document_metadata.file_name}")
+                print(f"Chunks created: {result.document_metadata.chunk_count}")
+        else:
+                print(f"Failed to process: {result.error_message}")
+
+    async def get_result(self, question):
+        self.inferencer.model.prompt_template = self.prompt_template
+        async for item in self.inferencer.infer_stream(query = question,
+                                    enable_reranking=False,
+                                    k=3):
+                yield item

space/space/app/rag/agents/gpt_customer_service_agent.py

@@ -0,0 +1,13 @@
+from rag.agents.agents import Agent
+from rag.pipeline.language_model import LM
+from rag.inference.inferencer import Inferencer
+
+class GPTCSAgent(Agent):
+    def __init__(self, inferencer : Inferencer , prompt_template):
+        super().__init__(inferencer, prompt_template)
+        self.inferencer = inferencer
+        self.prompt_template = prompt_template
+    async def get_result(self, question : str):
+        self.inferencer.model.prompt_template = self.prompt_template
+        print("Question received :", question)
+        return await self.inferencer.infer(query = question)

space/space/app/rag/agents/query_maker_agent.py

@@ -0,0 +1,13 @@
+from rag.agents.agents import Agent
+from rag.pipeline.language_model import LM
+from rag.inference.inferencer import Inferencer
+
+class QueryMakerAgent(Agent):
+    def __init__(self, inferencer : Inferencer , prompt_template):
+        super().__init__(inferencer, prompt_template)
+        self.inferencer = inferencer
+        self.prompt_template = prompt_template
+    async def get_result(self, question : str):
+        self.inferencer.model.prompt_template = self.prompt_template
+        print("Question received :", question)
+        return await self.inferencer.infer(query = question)

space/space/app/rag/chat_template/__init__.py

@@ -0,0 +1,29 @@
+def read_template_txt(file_path):
+    """Baca file txt biasa"""
+    with open(f"rag/chat_template/{file_path}.txt", 'r', encoding='utf-8') as f:
+        return f.read()
+def get_chat_template(file_name):
+    sys_prompt = read_template_txt(file_name)
+    return [
+        {
+            "role" : "system",
+            "content" : f"""
+            {sys_prompt}
+            """
+        },
+        {
+            "role" : "user",
+            "content" : """ 
+            
+            Please answer properly:  
+            {question} 
+
+            From given context :
+            {context}
+             
+            
+
+            """
+        }
+    ]
+    

space/space/app/rag/chat_template/customer_service.txt

@@ -0,0 +1,12 @@
+You are a friendly and professional Customer Service for Human Resource Information System (HRIS) field,
+representative, fluent in Indonesian. Your job is to assist customers with accurate information based on your company's basic knowledge. Follow these guidelines:
+
+- Always greet customers in a friendly and professional manner.
+- Your answers are contextual and objective.
+- Provide clear, easy-to-understand, and structured answers based on the context provided by the user.
+- If information is not available, offer alternative assistance or direct them to the appropriate channel.
+- Use polite language and empathize with the customer's needs.
+- Conclude by offering further assistance.
+- You are highly skilled in the area relevant to the given context.
+
+Please use the given context to answer accurately.

space/space/app/rag/chat_template/query_maker.txt

@@ -0,0 +1,35 @@
+Anda adalah agen AI yang tepat dan objektif, 
+Anda bertugas mengubah pertanyaan atau pernyataan pengguna menjadi query yang eksplisit dan efisien untuk keperluan pencarian dokumen dalam sistem RAG (Retrieval-Augmented Generation).
+
+Ikuti langkah-langkah berikut:
+
+1. Ekstrak bagian-bagian penting dari input pengguna:
+   - **Intent**: Tujuan utama atau jenis permintaan (misalnya: apa itu, cara, syarat, apakah bisa, berapa).
+   - **Entity/Noun Phrase**: Objek utama yang dibahas (misalnya: BPJS, tokenizer truncation, RWKV, gaji).
+   - **Context**: Informasi pendukung yang menyempitkan fokus (misalnya: kecelakaan kerja, gaji 1 juta per bulan, perusahaan mitra BPJS).
+   - **Question**: Pertanyaan spesifik yang ingin dijawab (misalnya: bagaimana prosesnya, apa manfaatnya, berapa jumlahnya).
+
+2. Setelah semua elemen diidentifikasi, bentuk **Query RAG** dengan struktur: [INTENT] + [ENTITY] + [CONTEXT] + [QUESTION]
+3. Gunakan bahasa natural yang ringkas, namun informatif dan eksplisit.
+4. Generate hanya hasil akhirnya saja berupa satu buah kalimat
+
+Contoh 0 :
+User Input:
+> Apa itu BPJS
+Output : Pengertian BPJS
+
+Contoh 1 :
+User Input:
+> Di mana lokasi PT Sakura System Solution ?
+
+Output: Lokasi PT Sakura System Solution
+
+Contoh 2:
+User Input:
+> Saya mengalami kecelakaan di kantor dan ingin tahu apakah bisa klaim BPJS karena perusahaan saya adalah mitra.
+
+Output: apakah bisa klaim BPJS kecelakaan kerja di kantor jika perusahaan mitra dan apakah saya memenuhi syarat
+
+
+**Tugas Anda sekarang:**
+Lakukan proses di atas untuk setiap input pengguna yang diberikan. Hasilkan query RAG akhir yang siap digunakan dalam pencarian dokumen.

space/space/app/rag/chat_template/query_maker_temp.txt

@@ -0,0 +1,30 @@
+Anda adalah agen AI yang tepat dan objektif, 
+Anda bertugas mengubah pertanyaan atau pernyataan pengguna menjadi query yang eksplisit dan efisien untuk keperluan pencarian dokumen dalam sistem RAG (Retrieval-Augmented Generation).
+
+Ikuti langkah-langkah berikut:
+
+1. Ekstrak bagian-bagian penting dari input pengguna:
+   - **Intent**: Tujuan utama atau jenis permintaan (misalnya: apa itu, cara, syarat, apakah bisa, berapa).
+   - **Entity/Noun Phrase**: Objek utama yang dibahas (misalnya: BPJS, tokenizer truncation, RWKV, gaji).
+   - **Context**: Informasi pendukung yang menyempitkan fokus (misalnya: kecelakaan kerja, gaji 1 juta per bulan, perusahaan mitra BPJS).
+   - **Question**: Pertanyaan spesifik yang ingin dijawab (misalnya: bagaimana prosesnya, apa manfaatnya, berapa jumlahnya).
+
+2. Setelah semua elemen diidentifikasi, bentuk **Query RAG** dengan struktur: [INTENT] + [ENTITY] + [CONTEXT] + [QUESTION]
+3. Gunakan bahasa natural yang ringkas, namun informatif dan eksplisit.
+4. Generate hanya hasil akhirnya saja berupa satu buah kalimat
+
+Contoh 1 :
+User Input:
+> Di mana lokasi PT Sakura System Solution ?
+
+Output: Lokasi PT Sakura System Solution
+
+Contoh 2:
+User Input:
+> Saya mengalami kecelakaan di kantor dan ingin tahu apakah bisa klaim BPJS karena perusahaan saya adalah mitra.
+
+Output: apakah bisa klaim BPJS kecelakaan kerja di kantor jika perusahaan mitra dan apakah saya memenuhi syarat
+
+
+**Tugas Anda sekarang:**
+Lakukan proses di atas untuk setiap input pengguna yang diberikan. Hasilkan query RAG akhir yang siap digunakan dalam pencarian dokumen.

space/space/app/rag/pipeline/language_model.py

@@ -0,0 +1,947 @@
+import torch
+import asyncio
+from transformers import AutoTokenizer, AutoModelForCausalLM, GenerationConfig, TextIteratorStreamer, BitsAndBytesConfig
+import torch
+from typing import Optional, Dict, Any, List, Union, Callable, Awaitable, AsyncGenerator
+import logging
+from dataclasses import dataclass
+from datetime import datetime
+from concurrent.futures import ThreadPoolExecutor
+from functools import partial
+from threading import Thread
+from rag.retriever.retriever_types import RetrievalResult
+from langchain_core.documents import Document
+import copy
+
+@dataclass
+class LMConfig:
+    model_name: str = "Qwen/Qwen2.5-1.5B-Instruct"
+    device: str = "cuda"
+    torch_dtype: torch.dtype = torch.float16
+    max_length: int = 2048
+    temperature: float = 0.7
+    top_p: float = 0.8
+    top_k: int = 50
+    do_sample: bool = True
+    quantization_config: any = None
+    pad_token_id: Optional[int] = None
+    eos_token_id: Optional[int] = None
+    # RAG-specific configs
+    max_context_length: int = 1500
+    context_separator: str = "\n---\n"
+    instruction_template: str = "system"  # "system", "instruction", "custom"
+    # Async-specific configs
+    max_workers: int = 2
+    generation_timeout: float = 30
+    repetition_penalty: float = 1.0
+    # Streaming-specific configs
+    stream_timeout: float = 100  # timeout untuk stream chunk
+    skip_prompt: bool = True     # skip prompt dari streaming output
+
+class LM:
+    """
+    Async LLM Qwen 0.5B dengan interface yang mudah digunakan
+    Termasuk prompt formatting khusus untuk RAG (Retrieval-Augmented Generation)
+    Dan support untuk text streaming
+    """
+    
+    def __init__(self, config: Optional[LMConfig] = None, prompt_template = [
+                 {"role": "system", "content": "You are a helpful assistant."},
+                 {"role": "user", "content": "{question}"}
+            ] ):
+        """
+        Inisialisasi LM
+        
+        Args:
+            config: Konfigurasi model (optional, akan menggunakan default jika None)
+        """
+        if(config is None):
+            self.config = LMConfig()
+        else:
+            self.config = config
+        self.tokenizer : AutoTokenizer = None
+        self.model = None
+        self.generation_config = None
+        self.is_loaded = False
+        self.executor = ThreadPoolExecutor(max_workers=self.config.max_workers)
+        self._lock = asyncio.Lock()
+        # Setup logging
+        logging.basicConfig(level=logging.INFO)
+        self.logger = logging.getLogger(__name__)
+        
+        # RAG prompt templates
+        self.prompt_template = prompt_template
+    
+    async def load_model(self) -> None:
+        """Load model dan tokenizer secara async"""
+        async with self._lock:
+            if self.is_loaded:
+                self.logger.info("Model already loaded")
+                return
+            
+            try:
+                self.logger.info(f"Loading model: {self.config.model_name}")
+                
+                # Load tokenizer dalam thread pool
+                self.tokenizer = await asyncio.get_event_loop().run_in_executor(
+                    self.executor,
+                    lambda: AutoTokenizer.from_pretrained(
+                        self.config.model_name,
+                        trust_remote_code=True,
+                        torch_dtype="auto",
+                        device_map="auto",
+                    )
+                )
+                
+                # Load model dalam thread pool
+                self.model = await asyncio.get_event_loop().run_in_executor(
+                    self.executor,
+                    lambda: AutoModelForCausalLM.from_pretrained(
+                        self.config.model_name,
+                        quantization_config=self.config.quantization_config,
+                        torch_dtype=self.config.torch_dtype,
+                        device_map=self.config.device,
+                        trust_remote_code=True
+                    )
+                )
+                
+                # Setup generation config
+                self.generation_config = GenerationConfig(
+                    max_length=self.config.max_length,
+                    temperature=self.config.temperature,
+                    top_p=self.config.top_p,
+                    top_k=self.config.top_k,
+                    do_sample=self.config.do_sample,
+                    pad_token_id=self.config.pad_token_id or self.tokenizer.eos_token_id,
+                    eos_token_id=self.config.eos_token_id or self.tokenizer.eos_token_id,
+                    repetition_penalty = self.config.repetition_penalty,
+                )
+                
+                self.is_loaded = True
+                self.logger.info("Model loaded successfully!")
+                
+            except Exception as e:
+                self.logger.error(f"Error loading model: {e}")
+                raise
+    
+    def get_available_templates(self) -> List[str]:
+        """
+        Dapatkan list template yang tersedia
+        
+        Returns:
+            List of available template names
+        """
+        return list(self.prompt_template)
+    
+    def preview_template(self, template_type: str, sample_question: str = "Apa itu AI?", 
+                        sample_context: str = "Artificial Intelligence adalah teknologi...") -> str:
+        """
+        Preview template dengan sample data
+        
+        Args:
+            template_type: Template type to preview
+            sample_question: Sample question
+            sample_context: Sample context
+            
+        Returns:
+            Preview of formatted template
+        """
+        if template_type not in self.prompt_template:
+            return f"Template '{template_type}' tidak tersedia. Available: {self.get_available_templates()}"
+        
+        template_data = copy.deepcopy(self.prompt_template)
+        # template_key = "user_template" if "user_template" in template_data else "template"
+        
+        return template_data["content"].format(
+            context=sample_context,
+            question=sample_question
+        )
+    
+    def _format_context(self, contexts: Union[List[str], RetrievalResult], numbering: bool = True) -> str:
+        """
+        Format retrieved contexts menjadi string yang coherent
+        
+        Args:
+            contexts: List of contexts (string atau RetrievalResult objects)
+            numbering: Whether to add document numbering
+            
+        Returns:
+            Formatted context string
+        """
+        if not contexts:
+            return ""
+        
+        formatted_contexts = []
+        self.logger.info(f"Context : {contexts}")
+        self.logger.info(f"Is RetrievalResult Contexts =  {isinstance(contexts, RetrievalResult)}")
+        if isinstance(contexts, RetrievalResult):
+                for i, ctx in enumerate(contexts.documents, 1):
+                    if numbering:
+                        header = f"[Dokumen {i}"
+                        if contexts.scores[i - 1]:
+                            header += f" (Skor: {contexts.scores[i - 1]:.3f})"
+                        header += "]"
+                    else:
+                        header = "[Dokumen"
+                        header += "]"
+                    formatted_contexts.append(f"{header}\n{ctx.page_content}")
+        else:
+            for i, ctx in enumerate(contexts, 1):
+                if isinstance(ctx, str):
+                    header = f"[Dokumen {i}]" if numbering else "[Dokumen]"
+                    formatted_contexts.append(f"{header}\n{ctx}")
+                else:
+                    header = f"[Dokumen {i}]" if numbering else "[Dokumen]"
+                    formatted_contexts.append(f"{header}\n{str(ctx)}")
+        
+        return self.config.context_separator.join(formatted_contexts)
+    
+    def _truncate_context(self, context: str, max_length: int) -> str:
+        """
+        Truncate context jika terlalu panjang
+        
+        Args:
+            context: Context string
+            max_length: Maximum length in characters
+            
+        Returns:
+            Truncated context
+        """
+        if len(context) <= max_length:
+            return context
+        
+        # Truncate dan tambahkan indicator
+        truncated = context[:max_length - 50]
+        return truncated + "\n\n[... Context dipotong karena terlalu panjang ...]"
+
+    async def format_rag_prompt(self, 
+                                question: str, 
+                                contexts: Union[List[str], RetrievalResult],
+                                template_type: Optional[str] = None,
+                                custom_template: Optional[str] = None,
+                                include_metadata: bool = True,
+                                context_numbering: bool = True,
+                                max_contexts: Optional[int] = None) -> str:
+        """
+        Format prompt untuk RAG dengan berbagai template options (async)
+        """
+        
+        def _format_sync():
+            
+            # Handle RetrievalResult secara eksplisit
+            if isinstance(contexts, RetrievalResult):
+                docs = contexts.documents
+                if max_contexts:
+                    docs = docs[:max_contexts]
+                processed_contexts = RetrievalResult(
+                    documents=docs,
+                    scores=contexts.scores[:len(docs)] if contexts.scores else [],
+                    query=contexts.query,
+                    retrieval_time=contexts.retrieval_time,
+                    metadata=contexts.metadata
+                )
+            else:
+                # contexts diasumsikan sebagai list biasa (list[str] atau list[Document])
+                processed_contexts = contexts[:max_contexts] if max_contexts and len(contexts) > max_contexts else contexts
+
+            # Format context menjadi string
+            formatted_context = self._format_context(processed_contexts, context_numbering)
+
+            # Truncate jika panjang melebihi batas
+            formatted_context = self._truncate_context(
+                formatted_context, 
+                self.config.max_context_length
+            )
+
+            # Tambah metadata jika diizinkan dan konteks adalah RetrievalResult
+            if include_metadata and isinstance(processed_contexts, RetrievalResult):
+                metadata_info = []
+                for i, doc in enumerate(processed_contexts.documents, 1):
+                    if hasattr(doc, "metadata") and doc.metadata:
+                        metadata_info.append(f"Dokumen {i}: {doc.metadata}")
+                # if metadata_info:
+                #     formatted_context += f"\n\n[Metadata]\n" + "\n".join(metadata_info)
+
+            return formatted_context
+
+        # Jalankan _format_sync di thread pool
+        formatted_context = await asyncio.get_event_loop().run_in_executor(
+            self.executor, _format_sync
+        )
+        self.logger.info(f"Formatted Context {formatted_context}")
+        # Tentukan template yang akan dipakai
+        if(template_type == ""):
+            self.config.instruction_template = "system"
+        # Gunakan custom template jika disediakan
+        if custom_template:
+            return custom_template.format(
+                context=formatted_context,
+                question=question
+            )
+        elif self.prompt_template:
+            print("question", question)
+           
+            template_data = copy.deepcopy(self.prompt_template)
+            print("template = ", template_type, "rag template = ", template_data)
+            # template_key = "user_template" if "user_template" in template_data else "template"
+
+            formatted_template = []
+            for cht in template_data:
+                    # Create a copy of the content to avoid modifying the original
+                content = cht["content"]
+                
+                # Format both placeholders at once to avoid KeyError
+                if "{context}" in content or "{question}" in content:
+                    try:
+                        content = content.format(
+                            context=formatted_context,
+                            question=question
+                        )
+                    except KeyError as e:
+                        self.logger.error(f"Missing placeholder in template: {e}")
+                        # Fallback: format only available placeholders
+                        if "{context}" in content:
+                            content = content.replace("{context}", formatted_context)
+                        if "{question}" in content:
+                            content = content.replace("{question}", question)
+                
+                # Create new dict with formatted content
+                formatted_chat = {
+                    "role": cht["role"],
+                    "content": content
+                }
+                
+                # Copy other fields if they exist
+                if "description" in cht:
+                    formatted_chat["description"] = cht["description"]
+                    
+                formatted_template.append(formatted_chat)
+
+            # self.logger.info(f"Formatted Template {formatted_template}")
+            # print("Forrmatted Template", formatted_template)
+            return formatted_template
+        else:
+            # Fallback default template
+            return [
+                 {"role": "system", "content": "You are a helpful assistant."},
+                 {"role": "user", "content": question}
+            ]
+
+    async def generate_stream(self, 
+                             prompt: List[Dict], 
+                             max_new_tokens: Optional[int] = None,
+                             temperature: Optional[float] = None,
+                             top_p: Optional[float] = None,
+                             **kwargs) -> AsyncGenerator[str, None]:
+        """
+        Generate text dari prompt secara streaming async
+        
+        Args:
+            prompt: Input text prompt
+            max_new_tokens: Maximum token baru yang akan di-generate
+            temperature: Temperature untuk generation (override config)
+            top_p: Top-p untuk generation (override config)
+            **kwargs: Parameter tambahan untuk generation
+            
+        Yields:
+            Generated text chunks
+        """
+        await self._check_model_loaded()
+        
+        # Setup streamer
+        streamer = TextIteratorStreamer(
+            self.tokenizer, 
+            timeout=self.config.stream_timeout,
+            skip_prompt=self.config.skip_prompt,
+            skip_special_tokens=True
+        )
+        
+        def _generate_sync():
+            try:
+                # Tokenize input
+                inputs = self.tokenizer.apply_chat_template(
+                    prompt,
+                    add_generation_prompt=True,
+                    return_tensors="pt"
+                )
+                
+                # Override generation config jika diperlukan
+                gen_config = self.generation_config
+                if any([max_new_tokens, temperature, top_p]):
+                    gen_config = GenerationConfig(
+                        max_new_tokens=max_new_tokens or self.config.max_length,
+                        temperature=temperature or self.config.temperature,
+                        top_p=top_p or self.config.top_p,
+                        top_k=self.config.top_k,
+                        do_sample=self.config.do_sample,
+                        pad_token_id=self.config.pad_token_id or self.tokenizer.eos_token_id,
+                        eos_token_id=self.config.eos_token_id or self.tokenizer.eos_token_id,
+                        repetition_penalty=self.config.repetition_penalty,
+                        **kwargs
+                    )
+                
+                # Move to GPU
+                self.model.to("cuda")
+                input_ids = inputs.to("cuda")
+                
+                # Generate dalam thread terpisah
+                generation_kwargs = {
+                    "input_ids": input_ids,
+                    "generation_config": gen_config,
+                    "streamer": streamer,
+                    **kwargs
+                }
+                
+                thread = Thread(target=self.model.generate, kwargs=generation_kwargs)
+                thread.start()
+                
+                return thread
+                
+            except Exception as e:
+                self.logger.error(f"Error during stream generation setup: {e}")
+                raise
+        
+        # Setup generation thread
+        generation_thread = await asyncio.get_event_loop().run_in_executor(
+            self.executor, _generate_sync
+        )
+        err = None
+        try:
+            # Stream tokens
+            for token in streamer:
+                if token:  # Skip empty tokens
+                    yield token
+                    
+            # Wait for generation thread to finish
+            err = await asyncio.get_event_loop().run_in_executor(
+                self.executor, generation_thread.join
+            )
+            
+        except Exception as e:
+            self.logger.error(f"Error during streaming: {e}, {err}")
+            # Make sure thread is cleaned up
+            if generation_thread.is_alive():
+                generation_thread.join(timeout=1.0)
+            raise
+
+    async def rag_generate_stream(self,
+                                 question: str,
+                                 contexts: Union[List[str], RetrievalResult],
+                                 template_type: Optional[str] = None,
+                                 max_new_tokens: Optional[int] = None,
+                                 temperature: Optional[float] = None,
+                                 **kwargs) -> AsyncGenerator[str, None]:
+        """
+        Generate jawaban untuk RAG secara streaming async
+        
+        Args:
+            question: User question
+            contexts: List of retrieved contexts
+            template_type: Template type untuk formatting
+            max_new_tokens: Maximum token baru yang akan di-generate
+            temperature: Temperature untuk generation
+            **kwargs: Parameter tambahan untuk generation
+            
+        Yields:
+            Generated answer chunks
+        """
+        await self._check_model_loaded()
+        
+        # Format prompt
+        prompt = await self.format_rag_prompt(question, contexts, template_type)
+        
+        # Generate dengan temperature yang lebih rendah untuk RAG (lebih faktual)
+        temp = temperature if temperature is not None else 0.3
+        
+        async for chunk in self.generate_stream(
+            prompt=prompt,
+            max_new_tokens=max_new_tokens,
+            temperature=temp,
+            **kwargs
+        ):
+            yield chunk
+
+    async def chat_stream(self, 
+                         messages: List[Dict[str, str]], 
+                         max_new_tokens: Optional[int] = None,
+                         **kwargs) -> AsyncGenerator[str, None]:
+        """
+        Chat dengan format conversation secara streaming async
+        
+        Args:
+            messages: List of messages dengan format [{"role": "user", "content": "..."}]
+            max_new_tokens: Maximum token baru yang akan di-generate
+            **kwargs: Parameter tambahan untuk generation
+            
+        Yields:
+            Response text chunks
+        """
+        await self._check_model_loaded()
+        
+        def _format_chat():
+            try:
+                # Format messages untuk chat
+                formatted_prompt = self.tokenizer.apply_chat_template(
+                    messages,
+                    tokenize=False,
+                    add_generation_prompt=True
+                )
+                return formatted_prompt
+                
+            except Exception as e:
+                self.logger.error(f"Error during chat formatting: {e}")
+                raise
+        
+        # Format chat template dalam thread pool
+        formatted_prompt = await asyncio.get_event_loop().run_in_executor(
+            self.executor, _format_chat
+        )
+        
+        async for chunk in self.generate_stream(
+            formatted_prompt, 
+            max_new_tokens=max_new_tokens,
+            **kwargs
+        ):
+            yield chunk
+
+    async def rag_chat_stream(self,
+                             messages: List[Dict[str, str]],
+                             contexts: Union[List[str], RetrievalResult],
+                             template_type: Optional[str] = None,
+                             max_new_tokens: Optional[int] = None,
+                             **kwargs) -> AsyncGenerator[str, None]:
+        """
+        RAG Chat dengan format conversation secara streaming async
+        
+        Args:
+            messages: List of messages dengan format [{"role": "user", "content": "..."}]
+            contexts: List of retrieved contexts
+            template_type: Template type untuk formatting
+            max_new_tokens: Maximum token baru yang akan di-generate
+            **kwargs: Parameter tambahan untuk generation
+            
+        Yields:
+            Response text chunks
+        """
+        await self._check_model_loaded()
+        
+        # Ambil last user message sebagai question
+        user_messages = [msg for msg in messages if msg.get("role") == "user"]
+        if not user_messages:
+            raise ValueError("No user message found in conversation")
+        
+        last_question = user_messages[-1]["content"]
+        
+        # Generate RAG response secara streaming
+        async for chunk in self.rag_generate_stream(
+            question=last_question,
+            contexts=contexts,
+            template_type=template_type,
+            max_new_tokens=max_new_tokens,
+            **kwargs
+        ):
+            yield chunk
+
+    # Utility method untuk collect full response dari stream
+    async def collect_stream(self, stream_generator: AsyncGenerator[str, None]) -> str:
+        """
+        Collect semua chunks dari stream generator menjadi full text
+        
+        Args:
+            stream_generator: AsyncGenerator yang menghasilkan text chunks
+            
+        Returns:
+            Complete generated text
+        """
+        chunks = []
+        async for chunk in stream_generator:
+            chunks.append(chunk)
+        return "".join(chunks)
+    
+    async def multi_template_generate(self,
+                                    question: str,
+                                    contexts: Union[List[str], RetrievalResult],
+                                    template_types: List[str],
+                                    max_new_tokens: Optional[int] = None,
+                                    **kwargs) -> Dict[str, str]:
+        """
+        Generate jawaban menggunakan multiple templates secara concurrent
+        
+        Args:
+            question: User question
+            contexts: List of retrieved contexts
+            template_types: List of template types to use
+            max_new_tokens: Maximum token baru yang akan di-generate
+            **kwargs: Parameter tambahan untuk generation
+            
+        Returns:
+            Dictionary dengan template_type sebagai key dan response sebagai value
+        """
+        await self._check_model_loaded()
+        
+        # Create tasks untuk concurrent generation
+        tasks = []
+        for template_type in template_types:
+            task = asyncio.create_task(
+                self._generate_single_template(
+                    question, contexts, template_type, max_new_tokens, **kwargs
+                )
+            )
+            tasks.append((template_type, task))
+        
+        # Wait for all tasks
+        results = {}
+        for template_type, task in tasks:
+            try:
+                response = await task
+                results[template_type] = response
+            except Exception as e:
+                self.logger.error(f"Error generating with template {template_type}: {e}")
+                results[template_type] = f"Error: {str(e)}"
+        
+        return results
+    
+    async def _generate_single_template(self,
+                                      question: str,
+                                      contexts: Union[List[str], RetrievalResult],
+                                      template_type: str,
+                                      max_new_tokens: Optional[int] = None,
+                                      **kwargs) -> str:
+        """Helper method untuk single template generation"""
+        return await self.rag_generate(
+            question=question,
+            contexts=contexts,
+            template_type=template_type,
+            max_new_tokens=max_new_tokens,
+            **kwargs
+        )
+    
+    async def rag_generate(self,
+                          question: str,
+                          contexts: Union[List[str], RetrievalResult],
+                          template_type: Optional[str] = None,
+                          max_new_tokens: Optional[int] = None,
+                          temperature: Optional[float] = None,
+                          **kwargs) -> str:
+        """
+        Generate jawaban untuk RAG secara async
+        
+        Args:
+            question: User question
+            contexts: List of retrieved contexts
+            template_type: Template type untuk formatting
+            max_new_tokens: Maximum token baru yang akan di-generate
+            temperature: Temperature untuk generation
+            **kwargs: Parameter tambahan untuk generation
+            
+        Returns:
+            Generated answer
+        """
+        await self._check_model_loaded()
+        
+        # Format prompt
+        prompt = await self.format_rag_prompt(question, contexts, template_type)
+        
+        # Generate dengan temperature yang lebih rendah untuk RAG (lebih faktual)
+        temp = temperature if temperature is not None else 0.3
+        
+        return await self.generate(
+            prompt=prompt,
+            max_new_tokens=max_new_tokens,
+            temperature=temp,
+            **kwargs
+        )
+    
+    async def rag_chat(self,
+                      messages: List[Dict[str, str]],
+                      contexts: Union[List[str], RetrievalResult],
+                      template_type: Optional[str] = None,
+                      max_new_tokens: Optional[int] = None,
+                      **kwargs) -> str:
+        """
+        RAG Chat dengan format conversation secara async
+        
+        Args:
+            messages: List of messages dengan format [{"role": "user", "content": "..."}]
+            contexts: List of retrieved contexts
+            template_type: Template type untuk formatting
+            max_new_tokens: Maximum token baru yang akan di-generate
+            **kwargs: Parameter tambahan untuk generation
+            
+        Returns:
+            Response text
+        """
+        await self._check_model_loaded()
+        
+        # Ambil last user message sebagai question
+        user_messages = [msg for msg in messages if msg.get("role") == "user"]
+        if not user_messages:
+            raise ValueError("No user message found in conversation")
+        
+        last_question = user_messages[-1]["content"]
+        
+        # Generate RAG response
+        return await self.rag_generate(
+            question=last_question,
+            contexts=contexts,
+            template_type=template_type,
+            max_new_tokens=max_new_tokens,
+            **kwargs
+        )
+    
+    async def _check_model_loaded(self) -> None:
+        """Cek apakah model sudah di-load secara async"""
+        if not self.is_loaded:
+            raise RuntimeError("Model belum di-load. Panggil await load_model() terlebih dahulu.")
+    
+    async def generate(self, 
+                      prompt: Union[List[Dict], str], 
+                      max_new_tokens: Optional[int] = None,
+                      temperature: Optional[float] = None,
+                      top_p: Optional[float] = None,
+                      **kwargs) -> str:
+        """
+        Generate text dari prompt secara async
+        
+        Args:
+            prompt: Input text prompt
+            max_new_tokens: Maximum token baru yang akan di-generate
+            temperature: Temperature untuk generation (override config)
+            top_p: Top-p untuk generation (override config)
+            **kwargs: Parameter tambahan untuk generation
+            
+        Returns:
+            Generated text
+        """
+        
+        await self._check_model_loaded()
+        
+        def _generate_sync():
+            try:
+                # Tokenize input
+                inputs = self.tokenizer.apply_chat_template(
+                    prompt,
+                    add_generation_prompt=True,
+                    return_tensors="pt"
+                )
+                
+                # Override generation config jika diperlukan
+                gen_config = self.generation_config
+                if any([max_new_tokens, temperature, top_p]):
+                    gen_config = GenerationConfig(
+                        max_new_tokens=max_new_tokens or self.config.max_length,
+                        temperature=temperature or self.config.temperature,
+                        top_p=top_p or self.config.top_p,
+                        top_k=self.config.top_k,
+                        do_sample=self.config.do_sample,
+                        pad_token_id=self.config.pad_token_id or self.tokenizer.eos_token_id,
+                        eos_token_id=self.config.eos_token_id or self.tokenizer.eos_token_id,
+                        repetition_penalty = self.config.repetition_penalty,
+                        **kwargs
+                    )
+                
+                # Generate
+                with torch.no_grad():
+                    
+                    self.model.to("cuda")
+                    input_ids = inputs.to("cuda")
+                    prompt_length = input_ids.shape[-1]
+                    outputs = self.model.generate(
+                        input_ids,
+                        generation_config=gen_config,
+                        **kwargs
+                    )
+                
+                # Decode output
+                generated_text = self.tokenizer.decode(
+                    outputs[0][prompt_length:], 
+                    skip_special_tokens=True
+                )
+
+                print("Generated Text", generated_text)
+                # Remove input prompt dari output
+                return generated_text
+                
+            except Exception as e:
+                self.logger.error(f"Error during generation: {e}")
+                raise
+        
+        # Run generation in thread pool dengan timeout
+        try:
+            result = await asyncio.wait_for(
+                asyncio.get_event_loop().run_in_executor(self.executor, _generate_sync),
+                timeout=self.config.generation_timeout
+            )
+            return result
+        except asyncio.TimeoutError:
+            self.logger.error(f"Generation timeout after {self.config.generation_timeout} seconds")
+            raise TimeoutError(f"Generation timeout after {self.config.generation_timeout} seconds")
+    
+    async def chat(self, 
+                  messages: List[Dict[str, str]], 
+                  max_new_tokens: Optional[int] = None,
+                  **kwargs) -> str:
+        """
+        Chat dengan format conversation secara async
+        
+        Args:
+            messages: List of messages dengan format [{"role": "user", "content": "..."}]
+            max_new_tokens: Maximum token baru yang akan di-generate
+            **kwargs: Parameter tambahan untuk generation
+            
+        Returns:
+            Response text
+        """
+        await self._check_model_loaded()
+        
+        def _format_chat():
+            try:
+                # Format messages untuk chat
+                formatted_prompt = self.tokenizer.apply_chat_template(
+                    messages,
+                    chat_template="rag",
+                    return_tensors="pt"
+                )
+                return formatted_prompt
+                
+            except Exception as e:
+                self.logger.error(f"Error during chat formatting: {e}")
+                raise
+        
+        # Format chat template dalam thread pool
+        formatted_prompt = await asyncio.get_event_loop().run_in_executor(
+            self.executor, _format_chat
+        )
+        
+        return await self.generate(
+            formatted_prompt, 
+            max_new_tokens=max_new_tokens,
+            **kwargs
+        )
+    
+    async def update_config(self, **kwargs) -> None:
+        """
+        Update konfigurasi model secara async
+        
+        Args:
+            **kwargs: Parameter konfigurasi yang akan diupdate
+        """
+        async with self._lock:
+            for key, value in kwargs.items():
+                if hasattr(self.config, key):
+                    setattr(self.config, key, value)
+                    self.logger.info(f"Updated {key} to {value}")
+                else:
+                    self.logger.warning(f"Unknown config parameter: {key}")
+            
+            # Update generation config jika model sudah loaded
+            if self.is_loaded:
+                self.generation_config = GenerationConfig(
+                    max_length=self.config.max_length,
+                    temperature=self.config.temperature,
+                    top_p=self.config.top_p,
+                    top_k=self.config.top_k,
+                    do_sample=self.config.do_sample,
+                    pad_token_id=self.config.pad_token_id or self.tokenizer.eos_token_id,
+                    eos_token_id=self.config.eos_token_id or self.tokenizer.eos_token_id,
+                    repetition_penalty = self.config.repetition_penalty,
+
+                )
+    
+    async def get_model_info(self) -> Dict[str, Any]:
+        """
+        Dapatkan informasi model secara async
+        
+        Returns:
+            Dictionary dengan informasi model
+        """
+        info = {
+            "model_name": self.config.model_name,
+            "is_loaded": self.is_loaded,
+            "config": self.config.__dict__
+        }
+        
+        if self.is_loaded:
+            # Get model info dalam thread pool
+            def _get_info():
+                return {
+                    "vocab_size": self.tokenizer.vocab_size,
+                    "model_parameters": sum(p.numel() for p in self.model.parameters()),
+                    "device": str(next(self.model.parameters()).device)
+                }
+            
+            model_info = await asyncio.get_event_loop().run_in_executor(
+                self.executor, _get_info
+            )
+            info.update(model_info)
+        
+        return info
+    
+    async def batch_generate(self, 
+                           prompts: List[str], 
+                           max_new_tokens: Optional[int] = None,
+                           **kwargs) -> List[str]:
+        """
+        Generate multiple prompts secara batch dan concurrent
+        
+        Args:
+            prompts: List of prompts to generate
+            max_new_tokens: Maximum token baru yang akan di-generate
+            **kwargs: Parameter tambahan untuk generation
+            
+        Returns:
+            List of generated texts
+        """
+        await self._check_model_loaded()
+        
+        # Create tasks untuk concurrent generation
+        tasks = [
+            asyncio.create_task(
+                self.generate(prompt, max_new_tokens=max_new_tokens, **kwargs)
+            )
+            for prompt in prompts
+        ]
+        
+        # Wait for all tasks
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+        
+        # Process results
+        processed_results = []
+        for i, result in enumerate(results):
+            if isinstance(result, Exception):
+                self.logger.error(f"Error generating prompt {i}: {result}")
+                processed_results.append(f"Error: {str(result)}")
+            else:
+                processed_results.append(result)
+        
+        return processed_results
+    
+    async def close(self) -> None:
+        """
+        Cleanup resources secara async
+        """
+        self.logger.info("Closing LM...")
+        
+        # Shutdown executor
+        self.executor.shutdown(wait=True)
+        
+        # Clear GPU memory
+        if hasattr(self, 'model') and self.model is not None:
+            del self.model
+        if hasattr(self, 'tokenizer') and self.tokenizer is not None:
+            del self.tokenizer
+        
+        if torch.cuda.is_available():
+            torch.cuda.empty_cache()
+        
+        self.is_loaded = False
+        self.logger.info("LM closed successfully")
+    
+    async def __aenter__(self):
+        """Async context manager entry"""
+        await self.load_model()
+        return self
+    
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        """Async context manager exit"""
+        await self.close()

space/space/app/rag/retriever/langchain_retriever.py

@@ -6,6 +6,7 @@ from langchain_openai import OpenAIEmbeddings
 
 # Vector stores
 from langchain_community.vectorstores import Chroma, FAISS, Pinecone
+from langchain.retrievers import EnsembleRetriever
 
 # Retriever base
 from langchain_core.vectorstores import VectorStoreRetriever
@@ -24,7 +25,6 @@ from langchain_core.documents import Document
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
-
 class LangChainRetriever(BaseRetriever):
     """LangChain-based retriever with multiple format support"""
 
@@ -160,17 +160,34 @@ class LangChainRetriever(BaseRetriever):
         except Exception as e:
             logger.error(f"Error adding documents: {str(e)}")
             return False
-
     async def _update_bm25_retriever(self, documents: List[Document]):
         try:
+            # Create BM25 retriever from documents
             self.bm25_retriever = BM25Retriever.from_documents(documents)
-            self.retriever = ContextualCompressionRetriever(
-                base_compressor=None,  # Optional: add compressor like CohereRerank or LLM-based
-                base_retriever=self.bm25_retriever  # Example: use BM25 as base, can combine
+            self.bm25_retriever.k = 10  # Set number of documents to retrieve
+            
+            # For hybrid search, you have several options:
+            
+            # Option 1: Use only BM25 retriever (simplest fix)
+            self.retriever = self.bm25_retriever
+            
+            vector_retriever = VectorStoreRetriever(
+                vectorstore=self.vectorstore,
+                search_kwargs={"k": 10}
+            )
+
+            self.retriever = EnsembleRetriever(
+                retrievers=[vector_retriever, self.bm25_retriever],
+                weights=[0.5, 0.5]  # Equal weight to both retrievers
             )
+            
         except Exception as e:
             logger.error(f"Error updating BM25 retriever: {str(e)}")
-
+            # Fallback to vector retriever only
+            self.retriever = VectorStoreRetriever(
+                vectorstore=self.vectorstore,
+                search_kwargs={"k": 10}
+            )
     async def retrieve(self, query: str, k: int = 5) -> RetrievalResult:
         try:
             import time
@@ -181,6 +198,7 @@ class LangChainRetriever(BaseRetriever):
                 None, self.retriever.get_relevant_documents, query
             )
             retrieved_docs = retrieved_docs[:k]
+
             scores = [0.9 - (i * 0.1) for i in range(len(retrieved_docs))]
 
             retrieval_time = time.time() - start_time
@@ -222,4 +240,4 @@ class LangChainRetriever(BaseRetriever):
         return list(self.processed_documents.values())
 
     def get_supported_formats(self) -> List[str]:
-        return self.document_loader.get_supported_extensions()
+        return self.document_loader.get_supported_extensions()

space/space/app/rtc/rtc_call_gpt.py

@@ -0,0 +1,364 @@
+import fastapi
+from fastapi.middleware.cors import CORSMiddleware
+
+from fastrtc import ReplyOnPause, Stream, AlgoOptions, SileroVadOptions, get_cloudflare_turn_credentials_async, get_cloudflare_turn_credentials
+from fastrtc.utils import audio_to_int16
+from openai import OpenAI
+from elevenlabs.client import ElevenLabs
+from dotenv import load_dotenv
+from tts.audio_edge_tts import EdgeTTS
+from rag import document_retriever
+import logging
+import time
+import platform
+import socket
+import os
+import numpy as np
+import io
+import wave
+import asyncio
+import librosa
+from pydub import AudioSegment
+# from stt.whisper_stt import WhisperSTT
+from collections import deque
+import torch
+import torchaudio.transforms as T
+import asyncio
+import concurrent.futures
+import threading
+from config.constant import HF_TOKEN
+import threading
+import re
+from openai import OpenAI
+from langchain_core.documents import Document
+
+from rag import ddgs
+# Load .env
+load_dotenv()
+logging.basicConfig(level=logging.INFO)
+
+class RTCHandler:
+    def __init__(self, openai_client: OpenAI,  whisper_stt = None, edge_tts : EdgeTTS = None):
+
+        """Initialize RTC handler with OpenAI, ElevenLabs, and EdgeTTS"""
+        self.whisper_stt = whisper_stt
+        self.edge_tts = edge_tts
+        self.prompt = ""
+        self.sys_prompt = """
+        
+        Kamu adalah customer service yang berbahasa Indonesia dengan baik sopan, santun, tapi santai pembawaannya.
+        Kamu bisa menjelaskan sesuatu secara baik dan membimbing customer dalam menghadapi masalah yang ada!
+
+        Kamu akan menjawab customer dengan media call /telepon jadi anda harus memberikan respon seperlunya saja
+        Tidak kepanjanngan, dan sangat jelas, 
+
+
+        Tidak lebih dari 50 kata.
+        """
+        self.openai_client = openai_client
+        self.messages = [
+            
+            {
+             "role": "system", 
+             "content": self.sys_prompt
+             }
+            
+            ]
+        self.full_response = ""
+        self.stream = None
+        self.app = None
+
+        self._setup_webrtc_ip()
+
+    def _setup_webrtc_ip(self):
+        """Setup WebRTC IP for Windows"""
+        if platform.system() == 'Windows':
+            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+            try:
+                s.connect(('8.8.8.8', 80))
+                local_ip = s.getsockname()[0]
+            except Exception:
+                local_ip = '127.0.0.1'
+            finally:
+                s.close()
+            os.environ['WEBRTC_IP'] = local_ip
+
+    def audio_to_bytes(self, audio_tuple, sample_rate=24000) -> io.BufferedReader:
+        sr, audio_data = audio_tuple
+        audio_int16 = audio_to_int16(audio_tuple)
+
+        buffer = io.BytesIO()
+        with wave.open(buffer, "wb") as wf:
+            wf.setnchannels(1)
+            wf.setsampwidth(2)
+            wf.setframerate(sr)
+            wf.writeframes(audio_int16.tobytes())
+        buffer.seek(0)
+        buffer.name = "audio.wav"
+        return buffer
+    def echo(self, audio):
+            """Process audio input and generate audio response - Optimized version"""
+            try:
+                stt_time = time.time()
+                logging.info("Performing STT")
+
+                
+                # transcription = self.whisper_stt.transcribe(self.audio_to_bytes(audio))
+                transcription = self.openai_client.audio.transcriptions.create(
+                    model="whisper-1",
+                    file=self.audio_to_bytes(audio),
+                    language="id"
+                )
+                
+                self.prompt = transcription.text
+                if self.prompt == "":
+                    logging.info("STT returned empty string")
+                    return
+                
+                logging.info(f"STT response: {transcription}")
+                
+                logging.info(f"STT took {time.time() - stt_time} seconds")
+
+                llm_time = time.time()
+                self.full_response = ""
+
+                # Single async function to handle both text streaming and audio generation
+                async def stream_text_to_audio():
+                    # self.prompt = "Perhitungan BPJS"
+                    retrieval_result = await document_retriever.retrieve(query = self.prompt)
+                    contexts = ""
+                    search_results = []
+                    
+                    async for result in ddgs.search(self.prompt, max_results=5):
+                        # self.logger.info(f"Processing SEO Result: {result[:100]}...")
+                        doc = Document(
+                            page_content=result,
+                            metadata={"source": "internet_search", "query": self.prompt}
+                        )
+                        print(doc)
+                        search_results.append(doc)
+            
+                    await document_retriever.add_documents([doc])
+
+                    i = 1
+                    for ctx in retrieval_result.documents:
+                        contexts += f"{i}. {ctx.page_content}" + "\n"
+                    print("Retrieved Contexts :", contexts)
+                    self.messages.append({"role": "user", "content": f""" 
+                                        Dari Konteks yang diberikan (jika diperlukan) :
+                                        {contexts}
+
+                                        Berikan jawaban atas pertanyaan yang diberikan :
+                                        {self.prompt}
+                                          
+                                          """})
+                    
+                    response = self.openai_client.chat.completions.create(
+                        model="gpt-3.5-turbo",
+                        messages=self.messages,
+                        max_tokens=200,
+                        stream=True
+                    )
+                    chunk_size = 1024
+                    no_buffer = 0
+                    text_buffer = ""
+                    
+                    for stream_data in response:
+                        print(stream_data.choices[0].delta.content)
+                        if stream_data.choices[0].finish_reason == "stop":
+                            if text_buffer:  # Yield sisa text
+                                yield text_buffer
+                            break
+                        if stream_data.choices[0].delta.content:
+                            chunk = stream_data.choices[0].delta.content
+                            self.full_response += chunk
+                            text_buffer += chunk
+                            # Generate audio immediately for each text chunk
+                            if re.search(r'[.,?;!]', chunk):
+                                try:
+                                    audio_buffer_gen =  await self.edge_tts.generate_audio_buffer(text_buffer)
+                                    audio_buffer = audio_buffer_gen[0]
+                                    
+                                    audio_buffer.seek(0)
+                                    
+                                    # Convert MP3 to PCM
+                                    audio_segment = AudioSegment.from_file(audio_buffer, format="mp3")
+                                    samples = np.array(audio_segment.get_array_of_samples()).astype(np.float32) / (2 ** 15)
+                                    
+                                    # Handle stereo to mono
+                                    if audio_segment.channels == 2:
+                                        samples = samples.reshape((-1, 2)).mean(axis=1)
+                                    
+                                    # # Resample to 24kHz
+                                    # resampled = librosa.resample(samples, orig_sr=audio_segment.frame_rate, target_sr=24000)
+                                    import torch
+                                    import torchaudio
+                                    
+                                    # Check if CUDA is available
+                                    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+                                    
+                                    # Convert numpy array to torch tensor and move to GPU
+                                    audio_tensor = torch.from_numpy(samples).unsqueeze(0).to(device)  # Add batch dimension and move to GPU
+                                    
+                                    # Create resampler and move to GPU
+                                    resampler = torchaudio.transforms.Resample(
+                                        orig_freq=audio_segment.frame_rate,
+                                        new_freq=24000
+                                    ).to(device)
+                                    
+                                    # Apply resampling on GPU
+                                    resampled_tensor = resampler(audio_tensor)
+                                    
+                                    # Convert back to numpy (move to CPU first)
+                                    resampled = resampled_tensor.squeeze(0).cpu().numpy()
+                                    # Yield audio chunks
+                                    for i in range(0, len(resampled), chunk_size):
+                                        yield (24000, resampled[i:i + chunk_size])
+                                    no_buffer = 0
+                                    text_buffer = ""
+                                except Exception as e:
+                                    logging.error(f"TTS generation failed for chunk: {e}")
+                                    continue
+                                    
+                        # elif stream_data["type"] == "metadata":
+                        #     setup_time = stream_data['data']['setup_time']
+                        #     print(f"\nSetup completed in {setup_time:.2f}s")
+                            
+                        # elif stream_data["type"] == "complete":
+                        #     total_time = stream_data['data']['total_time']
+                        #     print(f"\nTotal time: {total_time:.2f}s")
+                        #     break
+
+                # Run the single async function
+                loop = asyncio.new_event_loop()
+                asyncio.set_event_loop(loop)
+                
+                try:
+                    async_gen = stream_text_to_audio()
+                    while True:
+                        try:
+                            chunk = loop.run_until_complete(async_gen.__anext__())
+                            yield chunk
+                        except StopAsyncIteration:
+                            break
+                finally:
+                    loop.close()
+
+                self.messages.append({"role": "assistant", "content": self.full_response + " "})
+                logging.info(f"LLM response: {self.full_response}")
+                logging.info(f"LLM took {time.time() - llm_time} seconds")
+
+            except Exception as e:
+                logging.error(f"Error in echo function: {e}")
+                error_audio = np.zeros(24000, dtype=np.float32)
+                yield (24000, error_audio)
+    def reset_conversation(self):
+        logging.info("Resetting chat")
+        self.messages = [{"role": "system", "content": self.sys_prompt}]
+        self.full_response = ""
+
+    def create_stream(self):
+        try:
+            async def get_credentials():
+                return await get_cloudflare_turn_credentials_async(hf_token=HF_TOKEN)
+            self.stream = Stream(
+                rtc_configuration=get_credentials,
+                server_rtc_configuration=get_cloudflare_turn_credentials(ttl=360_000),
+                handler = ReplyOnPause(
+                    self.echo,
+                    algo_options=AlgoOptions(
+                        audio_chunk_duration=0.5,
+                        started_talking_threshold=0.1,
+                        speech_threshold=0.03
+                    ),
+                    model_options=SileroVadOptions(
+                        threshold=0.90,
+                        min_speech_duration_ms=250,
+                        min_silence_duration_ms=2000,
+                        speech_pad_ms=400,
+                        max_speech_duration_s=15
+                    )
+                ),
+                modality="audio",
+                mode="send-receive"
+            )
+            return self.stream
+        except Exception as e:
+            logging.error(f"Error creating stream: {e}")
+            raise
+
+    def create_fastapi_app(self):
+        try:
+            self.app = fastapi.FastAPI()
+            self.app.add_middleware(
+                CORSMiddleware,
+                allow_origins=["*"],
+                allow_credentials=True,
+                allow_methods=["*"],
+                allow_headers=["*"],
+            )
+
+            if not self.stream:
+                self.create_stream()
+            self.stream.mount(self.app)
+
+            @self.app.get("/reset")
+            async def reset():
+                try:
+                    self.reset_conversation()
+                    return {"status": "success"}
+                except Exception as e:
+                    logging.error(f"Error in reset endpoint: {e}")
+                    return {"status": "error", "message": str(e)}
+
+            @self.app.get("/status")
+            async def status():
+                try:
+                    return {
+                        "status": "running",
+                        "messages_count": len(self.messages),
+                        "last_response": self.full_response
+                    }
+                except Exception as e:
+                    logging.error(f"Error in status endpoint: {e}")
+                    return {"status": "error", "message": str(e)}
+
+            return self.app
+        except Exception as e:
+            logging.error(f"Error creating FastAPI app: {e}")
+            raise
+
+    def start_server(self, host: str = "0.0.0.0", port: int = 7860):
+        import uvicorn
+        if not self.app:
+            self.create_fastapi_app()
+        logging.info(f"Starting server on {host}:{port}")
+        try:
+            uvicorn.run(self.app, host=host, port=port, log_level="info")
+        except Exception as e:
+            logging.error(f"Error starting server: {e}")
+            raise
+    def launch_ui(self, browser: bool = True):
+        try:
+            if not self.stream:
+                self.create_stream()
+            if not self.app:
+                self.create_fastapi_app()
+            logging.info("Launching RTC UI...")
+            self.stream.ui.launch(self.app,
+                                  server_name="0.0.0.0",
+                                  server_port=7860,
+                                  )
+        except Exception as e:
+            logging.error(f"Error launching UI: {e}")
+            raise
+
+    def get_conversation_history(self):
+        return self.messages.copy()
+
+    def set_system_prompt(self, new_prompt: str):
+        self.sys_prompt = new_prompt
+        self.messages[0] = {"role": "system", "content": new_prompt}
+
+    def get_last_response(self):
+        return self.full_response

space/space/app/tests/qwen_llm_test.py

@@ -1,14 +1,14 @@
 from rag.retriever.retriever_types import *
-from rag.pipeline.qwen_llm import QwenLLM, QwenConfig
+from rag.pipeline.language_model import LM, LMConfig
 
 import warnings
 warnings.filterwarnings("ignore")
 
-async def test_qwen_llm():
+async def test_language_model():
     print(" ===== Testing QWEN LLM ==== ")
-    """Example usage of async QwenLLM"""
+    """Example usage of async LM"""
 
-    config = QwenConfig(
+    config = LMConfig(
         temperature=0.5,
         max_length=512,
         generation_timeout=30
@@ -23,20 +23,20 @@ async def test_qwen_llm():
     )
 
     # Using async context manager
-    async with QwenLLM(config) as llm:
+    async with LM(config) as llm:
           await test_qwen_single_generation(llm)
           await test_qwen_single_rag_generation(llm, contexts)
           await test_qwen_multiple_template_rag_generation(llm, contexts)
           await test_qwen_batch_generation(llm, contexts)
     print(" ===== Testing LLM DONE ==== ")
 
-async def test_qwen_single_generation(llm : QwenLLM):
+async def test_qwen_single_generation(llm : LM):
     print(" * Test Single Generation * ")
     response = await llm.generate("Jelaskan tentang AI")
     print(f"Response: {response}")
     print(" * Test Single Generation Done * ")
 
-async def test_qwen_single_rag_generation(llm : QwenLLM, ctx : RetrievalResult):
+async def test_qwen_single_rag_generation(llm : LM, ctx : RetrievalResult):
     print(" * Test Single RAG Generation * ")
     rag_response = await llm.rag_generate(
             question="Apa itu AI dan machine learning?",
@@ -46,7 +46,7 @@ async def test_qwen_single_rag_generation(llm : QwenLLM, ctx : RetrievalResult):
     print(f"RAG Response: {rag_response}")
     print(" * Test Single RAG Generation Done * ")
 
-async def test_qwen_multiple_template_rag_generation(llm : QwenLLM,ctx : RetrievalResult):
+async def test_qwen_multiple_template_rag_generation(llm : LM,ctx : RetrievalResult):
         print(" * Test Multiple Template Generation * ")
         multi_responses = await llm.multi_template_generate(
              question="Apa itu AI?",
@@ -57,7 +57,7 @@ async def test_qwen_multiple_template_rag_generation(llm : QwenLLM,ctx : Retriev
         print(" * Test Multiple Template Generation Done* ")
 
 
-async def test_qwen_batch_generation(llm : QwenLLM, ctx : RetrievalResult):
+async def test_qwen_batch_generation(llm : LM, ctx : RetrievalResult):
         print(" * Test Batch Generation * ")
         batch_responses = await llm.batch_generate([
              "Jelaskan tentang Python",

space/space/space/app/__chat__.py

@@ -1,13 +1,14 @@
 from tests.inference_test import test_inference
-
+from huggingface_hub import login
+login(new_session=False)
 import warnings
 warnings.filterwarnings("ignore")
 import asyncio
 def run_test():
     try:
         # await test_document_retriever()
-        # await test_qwen_llm()
-        asyncio.run(test_inference())
+        # await test_language_model()
+        test_inference()
     except Exception as e:
         print(e)
 

space/space/space/app/__test__.py

@@ -1,8 +1,3 @@
-
-# from tests.document_retriever_test import test_document_retriever
-# from tests.document_retriever_test import test_document_retriever
-# from tests.qwen_llm_test import test_qwen_llm
-# from tests.inference_test import test_inference
 from tests.rtc_test import test_rtc
 import warnings
 warnings.filterwarnings("ignore")

space/space/space/app/rag/__init__.py

@@ -1,17 +1,44 @@
-from rag.pipeline.qwen_llm import QwenLLM, QwenConfig
+from rag.pipeline.language_model import LM, LMConfig
 from rag.retriever.langchain_retriever import LangChainRetriever
 from rag.inference.inferencer import Inferencer, InferencerConfig
+from rag.agents.customer_service_agent import CSAgent
+from rag.agents.query_maker_agent import QueryMakerAgent
+from langchain_core.documents import Document
+from rag.web_search.duckduckgo_search import DuckDuckGoSearch
+from rag.chat_template import get_chat_template
+from transformers import BitsAndBytesConfig
+import torch
 
-config = QwenConfig(
+import logging
+import sys
+
+logging.basicConfig(
+    level=logging.DEBUG,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(funcName)s() - %(message)s',
+    handlers=[
+        logging.FileHandler('app.log'),
+        logging.StreamHandler(sys.stdout)
+    ]
+)
+bnb = BitsAndBytesConfig(
+                            load_in_4bit=True,                      # Enable 4-bit quantization
+                            bnb_4bit_use_double_quant=True,         # Use double quantization
+                            bnb_4bit_quant_type="nf4",              # Use NF4 quantization
+                            bnb_4bit_compute_dtype=torch.bfloat16,  # Compute dtype for 4bit base models
+        )
+
+
+config = LMConfig(
+                model_name = "Qwen/Qwen2.5-1.5B-Instruct",
                 temperature=0.3,
                 max_length=512,
-                generation_timeout=30,
+                generation_timeout=100,
                 repetition_penalty=1.1,
-                max_workers = 1,
-                do_sample = True,
-        )
-    
-llm = QwenLLM(
+                max_workers = 2,
+                quantization_config = bnb
+)
+
+llm = LM(
         config = config
 )
 
@@ -22,29 +49,42 @@ inferencer_config = InferencerConfig(
 )
 
 document_retriever = LangChainRetriever(
-        embedding_model="all-MiniLM-L6-v2",
+        embedding_model="sentence-transformers/all-MiniLM-L6-v2",
         vectorstore_type="chroma",
-        vectorstore_path="./vectorstore",
+        vectorstore_path="vectorstore/",
         use_hybrid_search=True,
         chunk_size=1000,
         chunk_overlap=200
 )
 
-inferencer = Inferencer(
+ddgs = DuckDuckGoSearch()
+
+cs_inferencer = Inferencer(
         model=llm,
         retriever=document_retriever,
+        # search_engine = ddgs,
         reranker=None,
         config=inferencer_config
 )
 
-async def get_response(question):
-    result = await inferencer.infer(question, "rag_response")
-    return result
+query_maker_inferencer = Inferencer(
+        model=llm,
+        config=inferencer_config
+)
+
+cs_agent = CSAgent(
+    inferencer = cs_inferencer,
+    prompt_template = get_chat_template("customer_service")
+)
+
+query_maker_chat_template = get_chat_template("query_maker")
+query_maker_chat_template[1]["content"] = """{question}"""
+
+query_maker_agent = QueryMakerAgent(
+    inferencer = query_maker_inferencer,
+    prompt_template = query_maker_chat_template
+)
+
+
+
 
-async def get_stream_response(question):
-    async for item in inferencer.infer_stream(query = question,
-                                             enable_reranking=False,
-                                             template_type="main_template",
-                                             k=3):
-            print("Stream Response :", item)
-            yield item

space/space/space/app/rag/inference/inferencer.py

@@ -1,6 +1,8 @@
 from rag.retriever.langchain_retriever import LangChainRetriever
-from rag.pipeline.qwen_llm import QwenLLM, QwenConfig
+from rag.pipeline.language_model import LM, LMConfig
 from rag.retriever.retriever_types import RetrievalResult
+from rag.web_search.duckduckgo_search import DuckDuckGoSearch
+from langchain_core.documents import Document
 # from rag.pipeline.reranker import BGEM3Reranker
 from typing import List, Union, Dict, Any, Optional, AsyncGenerator
 import asyncio
@@ -29,15 +31,16 @@ class Inferencer:
     """
     
     def __init__(self, 
-                 model: QwenLLM, 
-                 retriever: LangChainRetriever, 
+                 model: LM, 
+                 retriever: LangChainRetriever = None, 
+                 search_engine = None,
                  reranker=None,
                  config: Optional[InferencerConfig] = None):
         """
         Initialize Inferencer
         
         Args:
-            model: QwenLLM instance
+            model: LM instance
             retriever: LangChainRetriever instance
             reranker: Reranker instance (optional)
             config: InferencerConfig (optional)
@@ -45,6 +48,7 @@ class Inferencer:
         self.model = model
         self.retriever = retriever
         self.reranker = reranker
+        self.search_engine = search_engine
         self.config = config or InferencerConfig()
         
         # Setup logging
@@ -85,6 +89,7 @@ class Inferencer:
         try:
             start_time = datetime.now()
             contexts = await self.retriever.retrieve(query, k=k)
+            self.logger.info(f"Retrieved Contexts : {contexts}")
             retrieval_time = (datetime.now() - start_time).total_seconds()
             
             self.logger.info(f"Retrieved {len(contexts.documents) if hasattr(contexts, 'documents') else len(contexts)} contexts in {retrieval_time:.2f}s")
@@ -292,7 +297,7 @@ class Inferencer:
             yield chunk
     
     async def infer(self, 
-                   query: Union[str, List[str]], 
+                   query: str, 
                    response_type: Union[List[str], str] = None,
                    k: Optional[int] = None,
                    enable_reranking: Optional[bool] = None,
@@ -321,8 +326,12 @@ class Inferencer:
         
         try:
             # Step 1: Retrieve contexts
-            retrieved_contexts = await self.retrieve_context(main_query, k=k)
-            
+            if(self.search_engine):
+                await self.retrieve_from_search_engine(query, k = k)
+            if(self.retriever):
+                retrieved_contexts = await self.retrieve_context(main_query, k=k)
+            else:
+                retrieved_contexts  = ""
             # Step 2: Rerank contexts (if enabled)
             enable_rerank = enable_reranking if enable_reranking is not None else self.config.enable_reranking
             if enable_rerank:
@@ -363,7 +372,34 @@ class Inferencer:
         except Exception as e:
             self.logger.error(f"Error during inference: {e}")
             raise
-    
+    async def retrieve_from_search_engine(self, query: str, k: int = 3):
+        """
+        Alternative method: Process results as they come
+        """
+        from langchain_core.documents import Document
+        
+        search_results = []
+        
+        try:
+            # Process results one by one as they come
+            async for result in self.search_engine.search(query, max_results=k):
+                self.logger.info(f"Processing SEO Result: {result[:100]}...")
+                
+                doc = Document(
+                    page_content=result,
+                    metadata={"source": "internet_search", "query": query}
+                )
+                search_results.append(doc)
+                
+                # Optionally add to retriever immediately
+                await self.retriever.add_documents([doc])
+            
+            self.logger.info(f"Processed {len(search_results)} search results")
+            return search_results
+            
+        except Exception as e:
+            self.logger.error(f"Error in retrieve_from_search_engine_alternative: {e}", exc_info=True)
+            raise
     async def infer_stream(self, 
                           query: str,
                           k: Optional[int] = None,
@@ -389,8 +425,14 @@ class Inferencer:
         
         try:
             # Step 1: Retrieve contexts
-            retrieved_contexts = await self.retrieve_context(query, k=k)
+            if(self.search_engine):
+                await self.retrieve_from_search_engine(query, k = k)
+            if(self.retriever is not None):
+                retrieved_contexts = await self.retrieve_context(query, k=k)
+            else:
+                retrieved_contexts = ""
             
+
             # Step 2: Rerank contexts (if enabled)
             enable_rerank = enable_reranking if enable_reranking is not None else self.config.enable_reranking
             if enable_rerank:

space/space/space/app/rag/pipeline/qwen_llm.py

@@ -17,7 +17,7 @@ import copy
 @dataclass
 class QwenConfig:
     """Konfigurasi untuk model Qwen 0.5B"""
-    model_name: str = "Qwen/Qwen2.5-0.5B-Instruct"
+    model_name: str = "Qwen/Qwen2.5-1.5B-Instruct"
     device: str = "cuda"
     torch_dtype: torch.dtype = torch.float16
     max_length: int = 2048
@@ -286,14 +286,35 @@ class QwenLLM:
 
             formatted_template = []
             for cht in template_data:
-                # print("question for template = ", question)
-               
-                if("{context}" in cht["content"]):
-                    cht["content"] = cht["content"].format(context=formatted_context)
+                    # Create a copy of the content to avoid modifying the original
+                content = cht["content"]
+                
+                # Format both placeholders at once to avoid KeyError
+                if "{context}" in content or "{question}" in content:
+                    try:
+                        content = content.format(
+                            context=formatted_context,
+                            question=question
+                        )
+                    except KeyError as e:
+                        self.logger.error(f"Missing placeholder in template: {e}")
+                        # Fallback: format only available placeholders
+                        if "{context}" in content:
+                            content = content.replace("{context}", formatted_context)
+                        if "{question}" in content:
+                            content = content.replace("{question}", question)
+                
+                # Create new dict with formatted content
+                formatted_chat = {
+                    "role": cht["role"],
+                    "content": content
+                }
+                
+                # Copy other fields if they exist
+                if "description" in cht:
+                    formatted_chat["description"] = cht["description"]
                     
-                if("{question}" in cht["content"]):
-                    cht["content"] = cht["content"].format(question=question)
-                formatted_template.append(cht)
+                formatted_template.append(formatted_chat)
 
             self.logger.info("Formatted Template", formatted_template)
             print("Forrmatted Template", formatted_template)

space/space/space/app/rag/prompt_tuner/chat_template.py

@@ -8,18 +8,20 @@ def RAG_TEMPLATES():
 
             1. Selalu berikan sapaan yang ramah dan profesional
             2. Gunakan HANYA informasi dari knowledge base yang tersedia
-            3. Berikan jawaban yang jelas, mudah dipahami, dan terstruktur semuanya berdasarkan konteks yang diberikan yaitu :
-            {context}
+            3. Berikan jawaban yang jelas, mudah dipahami, dan terstruktur semuanya berdasarkan konteks yang diberikan user.
             4. Jika informasi tidak tersedia, tawarkan alternatif bantuan atau arahkan ke channel yang tepat
             5. Gunakan bahasa yang sopan dan empati terhadap kebutuhan pelanggan
             6. Akhiri dengan penawaran bantuan lebih lanjut
+            
             """,
             "description": "Template dengan system prompt untuk customer service professional"
             },
             {
             "role" : "user",
-            "content" : """
-            Dari konteks yang diberikan context berikan jawaban atas pertanyaan saya yaitu : {question}
+            "content" : """Dari konteks yang diberikan : {context} 
+            
+            berikan jawaban atas pertanyaan saya yaitu : {question}
+
             """
             },
         ],

space/space/space/app/rag/web_search/duckduckgo_search.py

@@ -0,0 +1,142 @@
+from ddgs import DDGS
+from langchain_community.document_loaders import AsyncChromiumLoader
+from langchain_community.document_transformers import BeautifulSoupTransformer
+import re
+import logging
+from typing import AsyncGenerator, List
+
+class DuckDuckGoSearch:
+    def __init__(self, html_loader: AsyncChromiumLoader = None, html_parser = None):
+        # Initialize dengan default values jika tidak diberikan
+        self.html_loader = html_loader or AsyncChromiumLoader([])
+        self.html_parser = html_parser or BeautifulSoupTransformer()
+        self.logger = logging.getLogger("ddgs_logger")
+    
+    async def get_page(self, urls: List[str]):
+        """Get page content from URLs - returns list of documents"""
+        try:
+            self.html_loader.urls = urls
+            html = await self.html_loader.aload()  # This returns a LIST
+            self.logger.info(f"search engine aload result: {len(html)} documents loaded")
+            
+            docs_transformed = self.html_parser.transform_documents(
+                html, 
+                tags_to_extract=["p"], 
+                remove_unwanted_tags=["a"]
+            )
+            return docs_transformed  # Returns LIST of documents
+            
+        except Exception as e:
+            self.logger.error(f"Error loading pages: {e}", exc_info=True)
+            return []  # Return empty list on error
+    
+    def truncate(self, text: str, max_words: int = 400) -> str:
+        """Truncate text to specified number of words"""
+        if not text:
+            return ""
+        
+        words = text.split()
+        if len(words) <= max_words:
+            return text
+            
+        truncated = " ".join(words[:max_words])
+        return truncated + "..." if len(words) > max_words else truncated
+    
+    async def search(self, query: str, max_results: int = 5) -> AsyncGenerator[str, None]:
+        """
+        Search and yield page contents one by one
+        
+        FIXED VERSION: Properly handle async iteration
+        """
+        try:
+            self.logger.info(f"Searching for: {query} (max_results: {max_results})")
+            
+            # Step 1: Get search results from DDGS (regular iterator)
+            results = DDGS().text(query, max_results=max_results)
+            urls = []
+            
+            # Step 2: Extract URLs using regular for loop (NOT async for)
+            for result in results:  # ← FIXED: Regular for loop
+                url = result.get('href')
+                if url:
+                    urls.append(url)
+            
+            self.logger.info(f"Found {len(urls)} URLs to process")
+            
+            if not urls:
+                self.logger.warning("No URLs found from search results")
+                return
+            
+            # Step 3: Get page content (await the coroutine first)
+            docs = await self.get_page(urls)  # ← FIXED: Await first, get list
+            
+            # Step 4: Process documents using regular for loop (NOT async for)
+            for doc in docs:  # ← FIXED: Regular for loop on list
+                try:
+                    if hasattr(doc, 'page_content') and doc.page_content:
+                        # Clean up text
+                        page_text = re.sub(r"\n\n+", "\n", doc.page_content)
+                        page_text = page_text.strip()
+                        
+                        if page_text:  # Only yield if there's actual content
+                            text = self.truncate(page_text)
+                            yield text  # Yield makes this an async generator
+                        
+                except Exception as e:
+                    self.logger.error(f"Error processing document: {e}")
+                    continue
+                    
+        except Exception as e:
+            self.logger.error(f"Error in search method: {e}", exc_info=True)
+            # Don't re-raise, just log and return (generator will be empty)
+    
+    async def search_with_metadata(self, query: str, max_results: int = 5) -> AsyncGenerator[dict, None]:
+        """
+        Alternative method that yields dictionaries with metadata
+        """
+        try:
+            results = DDGS().text(query, max_results=max_results)
+            urls_and_titles = []
+            
+            # Collect URLs and titles
+            for result in results:
+                url = result.get('href')
+                title = result.get('title', 'No title')
+                if url:
+                    urls_and_titles.append({'url': url, 'title': title})
+            
+            if not urls_and_titles:
+                return
+            
+            # Get page content
+            urls = [item['url'] for item in urls_and_titles]
+            docs = await self.get_page(urls)
+            
+            # Process and yield with metadata
+            for i, doc in enumerate(docs):
+                try:
+                    if hasattr(doc, 'page_content') and doc.page_content:
+                        page_text = re.sub(r"\n\n+", "\n", doc.page_content)
+                        page_text = page_text.strip()
+                        
+                        if page_text:
+                            text = self.truncate(page_text)
+                            
+                            # Get metadata if available
+                            metadata = {}
+                            if i < len(urls_and_titles):
+                                metadata = urls_and_titles[i]
+                            
+                            yield {
+                                'content': text,
+                                'url': metadata.get('url', 'Unknown'),
+                                'title': metadata.get('title', 'No title'),
+                                'word_count': len(text.split())
+                            }
+                            
+                except Exception as e:
+                    self.logger.error(f"Error processing document {i}: {e}")
+                    continue
+                    
+        except Exception as e:
+            self.logger.error(f"Error in search_with_metadata: {e}", exc_info=True)

space/space/space/app/rtc/__init__.py

@@ -2,11 +2,13 @@ from openai import OpenAI
 from elevenlabs.client import ElevenLabs
 from tts.audio_edge_tts import EdgeTTS
 from config.constant import OPENAI_API_KEY, ELEVENLABS_API_KEY
+# from rtc.rtc_call import RTCHandler
 from rtc.rtc_call import RTCHandler
 from stt.whisper_stt import WhisperSTT
 
-whisper_stt = WhisperSTT("turbo")
+whisper_stt = WhisperSTT(model_size = "base", device = "cuda")
 edge_tts = EdgeTTS("id-ID-ArdiNeural",  "+0%", "+0%")
+openai_client = OpenAI(api_key = OPENAI_API_KEY)
 rtc_handler = RTCHandler(whisper_stt, edge_tts)
 
 def handle_rtc():

space/space/space/app/rtc/rtc_call.py

@@ -30,7 +30,7 @@ import threading
 import re
 
 
-from rag import get_stream_response
+from rag import cs_agent
 # Load .env
 load_dotenv()
 logging.basicConfig(level=logging.INFO)
@@ -94,7 +94,7 @@ class RTCHandler:
                     logging.info("STT returned empty string")
                     return
 
-                logging.info(f"STT response: {prompt}")
+                logging.info(f"STT response: {transcription}")
                 self.messages.append({"role": "user", "content": prompt})
                 logging.info(f"STT took {time.time() - stt_time} seconds")
 
@@ -106,7 +106,7 @@ class RTCHandler:
                     chunk_size = 1024
                     no_buffer = 0
                     text_buffer = ""
-                    async for stream_data in get_stream_response(question=prompt):
+                    async for stream_data in cs_agent.get_result(question = prompt):
                         print(stream_data)
                         
                         if stream_data["type"] == "chunk":

space/space/space/app/stt/whisper_stt.py

@@ -1,31 +1,94 @@
+
 import whisper
+import torch
 from fastrtc.utils import audio_to_int16
 import io
 import os
 import tempfile
 
 class WhisperSTT:
-    def __init__(self, model_size: str = "base"):
+    def __init__(self, model_size: str = "base", device: str = "auto"):
         """
-        Initialize Whisper STT with specified model size (tiny, base, small, medium, large)
+        Initialize Whisper STT with specified model size and device
+        
+        Args:
+            model_size: Model size (tiny, base, small, medium, large)
+            device: Device to use ("auto", "cuda", "cpu")
         """
+        # Set up cache directory
         cache_dir = os.environ.get('WHISPER_CACHE_DIR', '/tmp/.cache/whisper')
         os.makedirs(cache_dir, exist_ok=True)
-        self.model = whisper.load_model(model_size, download_root=cache_dir)
+        
+        # Determine device
+        if device == "auto":
+            self.device = "cuda" if torch.cuda.is_available() else "cpu"
+        else:
+            self.device = device
+            
+        # Validate CUDA availability if requested
+        if self.device == "cuda" and not torch.cuda.is_available():
+            print("Warning: CUDA requested but not available. Falling back to CPU.")
+            self.device = "cpu"
+        
+        # Load model with device specification
+        print(f"Loading Whisper model '{model_size}' on device: {self.device}")
+        self.model = whisper.load_model(model_size, device=self.device, download_root=cache_dir)
         self.language = "id"  # ISO-639-1 code for Bahasa Indonesia
         
+        # Print GPU info if using CUDA
+        if self.device == "cuda":
+            gpu_name = torch.cuda.get_device_name(0)
+            gpu_memory = torch.cuda.get_device_properties(0).total_memory / 1024**3
+            print(f"Using GPU: {gpu_name} ({gpu_memory:.1f} GB)")
 
     def transcribe(self, audio: io.BufferedReader, language: str = "id") -> str:
-        # Simpan audio ke file sementara
+        """
+        Transcribe audio using Whisper
+        
+        Args:
+            audio: Audio file buffer
+            language: Language code (default: "id" for Indonesian)
+            
+        Returns:
+            Transcribed text
+        """
+        # Save audio to temporary file
         with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
             tmp.write(audio.read())
             tmp.flush()
             tmp_path = tmp.name
 
         try:
-            result = self.model.transcribe(tmp_path, language=language)
+            # Transcribe with GPU acceleration if available
+            result = self.model.transcribe(
+                tmp_path, 
+                language=language,
+                # Optional: Add fp16 for faster inference on supported GPUs
+                fp16=self.device == "cuda"
+            )
             return result.get("text", "")
         finally:
+            # Clean up temporary file
             os.remove(tmp_path)
-
-
+    
+    def get_device_info(self) -> dict:
+        """
+        Get information about the current device being used
+        
+        Returns:
+            Dictionary with device information
+        """
+        info = {
+            "device": self.device,
+            "cuda_available": torch.cuda.is_available()
+        }
+        
+        if self.device == "cuda" and torch.cuda.is_available():
+            info.update({
+                "gpu_name": torch.cuda.get_device_name(0),
+                "gpu_memory_gb": torch.cuda.get_device_properties(0).total_memory / 1024**3,
+                "gpu_memory_allocated_gb": torch.cuda.memory_allocated() / 1024**3,
+                "gpu_memory_reserved_gb": torch.cuda.memory_reserved() / 1024**3
+            })
+        
+        return info

space/space/space/app/tests/ddgs_test.py

@@ -0,0 +1,7 @@
+from rag.web_search import ddgs
+def test_ddgs():
+    # query = input()
+    # print("Searching for query = ", query)
+
+    print("*** searching result : **")
+    print(ddgs.search("Perhitungan uang lembur"))

space/space/space/app/tests/inference_test.py

@@ -1,69 +1,15 @@
 import gradio as gr
 import asyncio
-from rag.pipeline.qwen_llm import QwenLLM, QwenConfig
+from rag.pipeline.language_model import LM, LMConfig
 from rag.retriever.langchain_retriever import LangChainRetriever
 from rag.inference.inferencer import InferencerConfig, Inferencer
-
-async def test_inference():
+from rag import cs_agent, query_maker_agent
+def test_inference():
     """Main function that sets up and runs the RAG chatbot interface"""
     
     # Initialize RAG components
     print("==== Start Inference Test ===")
     
-    # Setup LLM
-    config = QwenConfig(
-        temperature=0.3,
-        max_length=512,
-        generation_timeout=30,
-        repetition_penalty=1.1,
-        do_sample = True,
-    )
-    
-    llm = QwenLLM(config=config)
-
-    # Setup Document Retriever
-    document_retriever = LangChainRetriever(
-        embedding_model="text-embedding-3-small",
-        vectorstore_type="chroma",
-        vectorstore_path="./vectorstore",
-        use_hybrid_search=True,
-        chunk_size=1000, 
-        chunk_overlap=200
-    )
-
-    # Load initial documents
-    file_paths = [
-        "../documents/bpjs.pdf",
-        "../documents/pph21.pdf",
-        "../documents/lembur.pdf",
-        "../documents/uu13.pdf",
-        "../documents/file.pdf",
-    ]
-    
-    for file_path in file_paths:
-        try:
-            result = await document_retriever.add_document_from_file(file_path)
-            if result.success:
-                print(f"Successfully processed: {result.document_metadata.file_name}")
-                print(f"Chunks created: {result.document_metadata.chunk_count}")
-            else:
-                print(f"Failed to process: {result.error_message}")
-        except Exception as e:
-            print(f"Error processing {file_path}: {e}")
-
-    # Setup Inferencer
-    inferencer_config = InferencerConfig(
-        default_k=2,
-        enable_reranking=False,
-        default_template_types=["system"]
-    )
-    
-    inferencer = Inferencer(
-        model=llm,
-        retriever=document_retriever,
-        reranker=None,
-        config=inferencer_config
-    )
     
     print("RAG system initialized successfully!")
 
@@ -73,16 +19,16 @@ async def test_inference():
             # Create new event loop for this thread
             loop = asyncio.new_event_loop()
             asyncio.set_event_loop(loop)
-            
+
             async def stream_response():
                 partial_response = ""
-                
-                async for stream_data in inferencer.infer_stream(
-                    query=message,
-                    k=3,
-                    template_type="main_template"
-                ):
-                    print(stream_data)
+                # print("message = ", message)
+                formatted_query = await query_maker_agent.get_result(question = message)
+                print("Formatted Query = ", formatted_query)
+                formatted_query = formatted_query['responses'][0]['rag_response']
+                await cs_agent.load_documents()
+                async for stream_data in cs_agent.get_result(question = formatted_query):
+
                     if stream_data["type"] == "chunk":
                         chunk = stream_data["data"]["chunk"]
                         partial_response += chunk
@@ -96,9 +42,8 @@ async def test_inference():
                         total_time = stream_data['data']['total_time']
                         print(f"\nTotal time: {total_time:.2f}s")
             
-            # Execute async generator
             async_gen = stream_response()
-            
+
             try:
                 while True:
                     result = loop.run_until_complete(async_gen.__anext__())
@@ -121,7 +66,7 @@ async def test_inference():
             asyncio.set_event_loop(loop)
             
             async def add_doc():
-                result = await document_retriever.add_document_from_file(file_path)
+                result = ""
                 return result
             
             result = loop.run_until_complete(add_doc())
@@ -158,8 +103,7 @@ async def test_inference():
     # Membuat interface Gradio
     with gr.Blocks(css=css, title="RAG Chatbot") as demo:
         gr.Markdown("""
-        # 🤖 RAG Chatbot dengan Text Streaming
-        Chatbot berbasis Retrieval-Augmented Generation (RAG) dengan dukungan streaming response.
+        # 🤖 SakuraAI, Virtual Assistant 
         """)
         
         # Status indicator

space/space/space/space/space/.env.example

@@ -0,0 +1,3 @@
+OPENAI_API_KEY =
+ELEVENLABS_API_KEY =
+HF_TOKEN = 

space/space/space/space/space/.gitattributes

@@ -0,0 +1,37 @@
+*.7z filter=lfs diff=lfs merge=lfs -text
+*.arrow filter=lfs diff=lfs merge=lfs -text
+*.bin filter=lfs diff=lfs merge=lfs -text
+*.bz2 filter=lfs diff=lfs merge=lfs -text
+*.ckpt filter=lfs diff=lfs merge=lfs -text
+*.ftz filter=lfs diff=lfs merge=lfs -text
+*.gz filter=lfs diff=lfs merge=lfs -text
+*.h5 filter=lfs diff=lfs merge=lfs -text
+*.joblib filter=lfs diff=lfs merge=lfs -text
+*.lfs.* filter=lfs diff=lfs merge=lfs -text
+*.mlmodel filter=lfs diff=lfs merge=lfs -text
+*.model filter=lfs diff=lfs merge=lfs -text
+*.msgpack filter=lfs diff=lfs merge=lfs -text
+*.npy filter=lfs diff=lfs merge=lfs -text
+*.npz filter=lfs diff=lfs merge=lfs -text
+*.onnx filter=lfs diff=lfs merge=lfs -text
+*.ot filter=lfs diff=lfs merge=lfs -text
+*.parquet filter=lfs diff=lfs merge=lfs -text
+*.pb filter=lfs diff=lfs merge=lfs -text
+*.pickle filter=lfs diff=lfs merge=lfs -text
+*.pkl filter=lfs diff=lfs merge=lfs -text
+*.pt filter=lfs diff=lfs merge=lfs -text
+*.pth filter=lfs diff=lfs merge=lfs -text
+*.rar filter=lfs diff=lfs merge=lfs -text
+*.safetensors filter=lfs diff=lfs merge=lfs -text
+saved_model/**/* filter=lfs diff=lfs merge=lfs -text
+*.tar.* filter=lfs diff=lfs merge=lfs -text
+*.tar filter=lfs diff=lfs merge=lfs -text
+*.tflite filter=lfs diff=lfs merge=lfs -text
+*.tgz filter=lfs diff=lfs merge=lfs -text
+*.wasm filter=lfs diff=lfs merge=lfs -text
+*.xz filter=lfs diff=lfs merge=lfs -text
+*.zip filter=lfs diff=lfs merge=lfs -text
+*.zst filter=lfs diff=lfs merge=lfs -text
+*tfevents* filter=lfs diff=lfs merge=lfs -text
+documents/SPISy[[:space:]]SaaS[[:space:]]To[[:space:]]The[[:space:]]Next[[:space:]]Level.pdf filter=lfs diff=lfs merge=lfs -text
+documents/file.pdf filter=lfs diff=lfs merge=lfs -text

space/space/space/space/space/.github/workflows/deploy-to-huggingface.yml

@@ -0,0 +1,52 @@
+name: Deploy to Huggingface
+
+on:
+  push:
+    branches:
+      - main
+
+jobs:
+  deploy-to-huggingface:
+    runs-on: ubuntu-latest
+
+    steps:
+      # Checkout repository
+      - name: Checkout Repository
+        uses: actions/checkout@v3
+
+      # Setup Git
+      - name: Setup Git for Huggingface
+        run: |
+          git config --global user.email "abdan.hafidz@gmail.com"
+          git config --global user.name "abdanhafidz"
+
+      # Clone Huggingface Space Repository
+      - name: Clone Huggingface Space
+        env:
+          HF_TOKEN: ${{ secrets.HF_TOKEN }}
+        run: |
+          git clone https://huggingface.co/spaces/lifedebugger/cs-ai-sakura-dev space
+
+      # Update Git Remote URL and Pull Latest Changes
+      - name: Update Remote and Pull Changes
+        env:
+          HF_TOKEN: ${{ secrets.HF_TOKEN }}
+        run: |
+          cd space
+          git remote set-url origin https://lifedebugger:$HF_TOKEN@huggingface.co/spaces/lifedebugger/cs-ai-sakura-dev
+          git pull origin main || echo "No changes to pull"
+
+      # Copy Files to Huggingface Space
+      - name: Copy Files to Space
+        run: |
+          rsync -av --exclude='.git' ./ space/
+
+      # Commit and Push to Huggingface Space
+      - name: Commit and Push to Huggingface
+        env:
+          HF_TOKEN: ${{ secrets.HF_TOKEN }}
+        run: |
+          cd space
+          git add .
+          git commit -m "Deploy files from GitHub repository" || echo "No changes to commit"
+          git push origin main || echo "No changes to push"

space/space/space/space/space/.gitignore

@@ -0,0 +1,9 @@
+.venv/
+venv/
+.vscode/
+__pycache__/
+my_vectorstore/
+FlagEmbedding/
+.env
+vectorstore/
+documents/

space/space/space/space/space/Dockerfile

@@ -0,0 +1,49 @@
+# Gunakan image dasar Python versi 3.13
+FROM python:3.13
+
+# Tambahkan user non-root untuk keamanan
+RUN useradd -m -u 1001 appuser
+
+# Set working directory
+WORKDIR /rag_be
+
+# Set cache directories ke writable location
+ENV HF_HOME=/tmp/.cache/huggingface
+ENV TRANSFORMERS_CACHE=/tmp/.cache/transformers
+ENV TORCH_HOME=/tmp/.cache/torch
+ENV XDG_CACHE_HOME=/tmp/.cache
+ENV TMPDIR=/tmp
+ENV WHISPER_CACHE_DIR=/tmp/.cache/whisper
+
+# Copy requirements dan install dependencies
+COPY requirements.txt ./
+RUN pip install --no-cache-dir --upgrade -r requirements.txt
+
+# Copy aplikasi dengan ownership ke appuser
+COPY --chown=appuser:appuser . /rag_be
+
+# Buat file .env dengan variabel environment menggunakan Hugging Face secrets
+RUN --mount=type=secret,id=OPENAI_API_KEY,mode=0444,required=false \
+    --mount=type=secret,id=HF_TOKEN,mode=0444,required=false \
+    --mount=type=secret,id=ELEVENLABS_API_KEY,mode=0444,required=false \
+    echo "OPENAI_API_KEY=$(cat /run/secrets/OPENAI_API_KEY 2>/dev/null || echo '')" >> .env && \
+    echo "HF_TOKEN=$(cat /run/secrets/HF_TOKEN 2>/dev/null || echo '')" >> .env && \
+    echo "ELEVENLABS_API_KEY=$(cat /run/secrets/ELEVENLABS_API_KEY 2>/dev/null || echo '')" >> .env
+    
+RUN ls -l /rag_be/app && whoami && id
+
+# Buat directories yang diperlukan dengan permissions yang tepat
+RUN mkdir -p /tmp/.cache /tmp/.cache/whisper /tmp/.cache/huggingface /rag_be/vectorstore  /tmp/.cache/transformers /tmp/.cache/torch \
+             /rag_be/app/vectorstore /rag_be/documents  && \
+    chmod -R 777 /tmp/.cache /rag_be/app /rag_be/app/vectorstore /rag_be/vectorstore /rag_be/documents && \
+    chown -R appuser:appuser /tmp/.cache /rag_be/app /rag_be/app/vectorstore /rag_be/vectorstore  /rag_be/documents /rag_be/.env
+
+RUN apt-get update && apt-get install -y ffmpeg
+# Beralih ke user non-root
+USER appuser
+
+# Expose port untuk Hugging Face Spaces
+EXPOSE 7860
+
+# Jalankan aplikasi
+CMD ["python", "app/__test__.py"]

space/space/space/space/space/README.md

@@ -0,0 +1,31 @@
+---
+title: Cs Ai Sakura Dev
+emoji: 🏢
+colorFrom: indigo
+colorTo: indigo
+sdk: docker
+pinned: false
+---
+
+**Install The Requirements**
+
+1.Create a virtual environment and install the dependencies
+```
+python3 -m venv env
+source env/bin/activate
+pip install -r requirements.txt
+```
+
+2. Set your OPENAI_API_KEY in .env file
+
+3. **TO LAUNCH THE GRADIO UI** Run the command below :
+```
+cd app
+python __test__.py
+```
+
+4. **TO LAUNCH THE API ENDPOINT (SERVER)** Run the command below :
+```
+cd app
+python __server__.py
+```

space/space/space/space/space/app/.gradio/certificate.pem

@@ -0,0 +1,31 @@
+-----BEGIN CERTIFICATE-----
+MIIFazCCA1OgAwIBAgIRAIIQz7DSQONZRGPgu2OCiwAwDQYJKoZIhvcNAQELBQAw
+TzELMAkGA1UEBhMCVVMxKTAnBgNVBAoTIEludGVybmV0IFNlY3VyaXR5IFJlc2Vh
+cmNoIEdyb3VwMRUwEwYDVQQDEwxJU1JHIFJvb3QgWDEwHhcNMTUwNjA0MTEwNDM4
+WhcNMzUwNjA0MTEwNDM4WjBPMQswCQYDVQQGEwJVUzEpMCcGA1UEChMgSW50ZXJu
+ZXQgU2VjdXJpdHkgUmVzZWFyY2ggR3JvdXAxFTATBgNVBAMTDElTUkcgUm9vdCBY
+MTCCAiIwDQYJKoZIhvcNAQEBBQADggIPADCCAgoCggIBAK3oJHP0FDfzm54rVygc
+h77ct984kIxuPOZXoHj3dcKi/vVqbvYATyjb3miGbESTtrFj/RQSa78f0uoxmyF+
+0TM8ukj13Xnfs7j/EvEhmkvBioZxaUpmZmyPfjxwv60pIgbz5MDmgK7iS4+3mX6U
+A5/TR5d8mUgjU+g4rk8Kb4Mu0UlXjIB0ttov0DiNewNwIRt18jA8+o+u3dpjq+sW
+T8KOEUt+zwvo/7V3LvSye0rgTBIlDHCNAymg4VMk7BPZ7hm/ELNKjD+Jo2FR3qyH
+B5T0Y3HsLuJvW5iB4YlcNHlsdu87kGJ55tukmi8mxdAQ4Q7e2RCOFvu396j3x+UC
+B5iPNgiV5+I3lg02dZ77DnKxHZu8A/lJBdiB3QW0KtZB6awBdpUKD9jf1b0SHzUv
+KBds0pjBqAlkd25HN7rOrFleaJ1/ctaJxQZBKT5ZPt0m9STJEadao0xAH0ahmbWn
+OlFuhjuefXKnEgV4We0+UXgVCwOPjdAvBbI+e0ocS3MFEvzG6uBQE3xDk3SzynTn
+jh8BCNAw1FtxNrQHusEwMFxIt4I7mKZ9YIqioymCzLq9gwQbooMDQaHWBfEbwrbw
+qHyGO0aoSCqI3Haadr8faqU9GY/rOPNk3sgrDQoo//fb4hVC1CLQJ13hef4Y53CI
+rU7m2Ys6xt0nUW7/vGT1M0NPAgMBAAGjQjBAMA4GA1UdDwEB/wQEAwIBBjAPBgNV
+HRMBAf8EBTADAQH/MB0GA1UdDgQWBBR5tFnme7bl5AFzgAiIyBpY9umbbjANBgkq
+hkiG9w0BAQsFAAOCAgEAVR9YqbyyqFDQDLHYGmkgJykIrGF1XIpu+ILlaS/V9lZL
+ubhzEFnTIZd+50xx+7LSYK05qAvqFyFWhfFQDlnrzuBZ6brJFe+GnY+EgPbk6ZGQ
+3BebYhtF8GaV0nxvwuo77x/Py9auJ/GpsMiu/X1+mvoiBOv/2X/qkSsisRcOj/KK
+NFtY2PwByVS5uCbMiogziUwthDyC3+6WVwW6LLv3xLfHTjuCvjHIInNzktHCgKQ5
+ORAzI4JMPJ+GslWYHb4phowim57iaztXOoJwTdwJx4nLCgdNbOhdjsnvzqvHu7Ur
+TkXWStAmzOVyyghqpZXjFaH3pO3JLF+l+/+sKAIuvtd7u+Nxe5AW0wdeRlN8NwdC
+jNPElpzVmbUq4JUagEiuTDkHzsxHpFKVK7q4+63SM1N95R1NbdWhscdCb+ZAJzVc
+oyi3B43njTOQ5yOf+1CceWxG1bQVs5ZufpsMljq4Ui0/1lvh+wjChP4kqKOJ2qxq
+4RgqsahDYVvTH9w7jXbyLeiNdd8XM2w9U/t7y0Ff/9yi0GE44Za4rF2LN9d11TPA
+mRGunUHBcnWEvgJBQl9nJEiU0Zsnvgc/ubhPgXRR4Xq37Z0j4r7g1SgEEzwxA57d
+emyPxgcYxn/eR44/KJ4EBs+lVDR3veyJm+kXQ99b21/+jh5Xos1AnX5iItreGCc=
+-----END CERTIFICATE-----

space/space/space/space/space/app/__chat__.py

@@ -0,0 +1,14 @@
+from tests.inference_test import test_inference
+
+import warnings
+warnings.filterwarnings("ignore")
+import asyncio
+def run_test():
+    try:
+        # await test_document_retriever()
+        # await test_qwen_llm()
+        asyncio.run(test_inference())
+    except Exception as e:
+        print(e)
+
+run_test()

space/space/space/space/space/app/__server__.py

@@ -0,0 +1,3 @@
+import rtc
+
+rtc.handle_rtc_server()

space/space/space/space/space/app/__test__.py

@@ -0,0 +1,19 @@
+
+# from tests.document_retriever_test import test_document_retriever
+# from tests.document_retriever_test import test_document_retriever
+# from tests.qwen_llm_test import test_qwen_llm
+# from tests.inference_test import test_inference
+from tests.rtc_test import test_rtc
+import warnings
+warnings.filterwarnings("ignore")
+import asyncio
+def run_test():
+    try:
+        # await test_document_retriever()
+        # await test_qwen_llm()
+        # asyncio.run(test_inference())
+        test_rtc()
+    except Exception as e:
+        print(e)
+
+run_test()

space/space/space/space/space/app/config/constant.py

@@ -0,0 +1,7 @@
+from dotenv import load_dotenv
+import os
+load_dotenv()
+
+OPENAI_API_KEY  = os.getenv('OPENAI_API_KEY')
+ELEVENLABS_API_KEY = os.getenv('ELEVENLABS_API_KEY')
+HF_TOKEN = os.getenv("HF_TOKEN")

space/space/space/space/space/app/rag/__init__.py

@@ -0,0 +1,50 @@
+from rag.pipeline.qwen_llm import QwenLLM, QwenConfig
+from rag.retriever.langchain_retriever import LangChainRetriever
+from rag.inference.inferencer import Inferencer, InferencerConfig
+
+config = QwenConfig(
+                temperature=0.3,
+                max_length=512,
+                generation_timeout=30,
+                repetition_penalty=1.1,
+                max_workers = 1,
+                do_sample = True,
+        )
+    
+llm = QwenLLM(
+        config = config
+)
+
+inferencer_config = InferencerConfig(
+        default_k=5,
+        enable_reranking=False,
+        default_template_types="main_template"
+)
+
+document_retriever = LangChainRetriever(
+        embedding_model="all-MiniLM-L6-v2",
+        vectorstore_type="chroma",
+        vectorstore_path="./vectorstore",
+        use_hybrid_search=True,
+        chunk_size=1000,
+        chunk_overlap=200
+)
+
+inferencer = Inferencer(
+        model=llm,
+        retriever=document_retriever,
+        reranker=None,
+        config=inferencer_config
+)
+
+async def get_response(question):
+    result = await inferencer.infer(question, "rag_response")
+    return result
+
+async def get_stream_response(question):
+    async for item in inferencer.infer_stream(query = question,
+                                             enable_reranking=False,
+                                             template_type="main_template",
+                                             k=3):
+            print("Stream Response :", item)
+            yield item

space/space/space/space/space/app/rag/inference/inferencer.py

@@ -0,0 +1,552 @@
+from rag.retriever.langchain_retriever import LangChainRetriever
+from rag.pipeline.qwen_llm import QwenLLM, QwenConfig
+from rag.retriever.retriever_types import RetrievalResult
+# from rag.pipeline.reranker import BGEM3Reranker
+from typing import List, Union, Dict, Any, Optional, AsyncGenerator
+import asyncio
+import logging
+from dataclasses import dataclass
+from datetime import datetime
+
+@dataclass
+class InferencerConfig:
+    """Konfigurasi untuk Inferencer"""
+    default_k: int = 5
+    max_contexts: int = 10
+    enable_reranking: bool = False
+    reranker_top_k: int = 5
+    default_template_types: List[str] = None
+    enable_logging: bool = True
+    response_timeout: float = 30.0
+    
+    def __post_init__(self):
+        if self.default_template_types is None:
+            self.default_template_types = ["system", "instruction", "friendly"]
+
+class Inferencer:
+    """
+    Advanced RAG Inferencer dengan support untuk streaming, reranking, dan multiple response types
+    """
+    
+    def __init__(self, 
+                 model: QwenLLM, 
+                 retriever: LangChainRetriever, 
+                 reranker=None,
+                 config: Optional[InferencerConfig] = None):
+        """
+        Initialize Inferencer
+        
+        Args:
+            model: QwenLLM instance
+            retriever: LangChainRetriever instance
+            reranker: Reranker instance (optional)
+            config: InferencerConfig (optional)
+        """
+        self.model = model
+        self.retriever = retriever
+        self.reranker = reranker
+        self.config = config or InferencerConfig()
+        
+        # Setup logging
+        if self.config.enable_logging:
+            logging.basicConfig(level=logging.INFO)
+            self.logger = logging.getLogger(__name__)
+        else:
+            self.logger = logging.getLogger(__name__)
+            self.logger.setLevel(logging.ERROR)
+        
+        # Model loading flag
+        self._model_loaded = False
+    
+    async def _ensure_model_loaded(self):
+        """Pastikan model sudah diload (hanya sekali)"""
+        if not self._model_loaded:
+            self.logger.info("Loading model...")
+            await self.model.load_model()
+            self._model_loaded = True
+            self.logger.info("Model loaded successfully")
+    
+    async def retrieve_context(self, 
+                             query: str, 
+                             k: Optional[int] = None) -> RetrievalResult:
+        """
+        Retrieve context documents
+        
+        Args:
+            query: Search query
+            k: Number of documents to retrieve
+            
+        Returns:
+            RetrievalResult object
+        """
+        k = k or self.config.default_k
+        self.logger.info(f"Retrieving {k} contexts for query: {query[:50]}...")
+        
+        try:
+            start_time = datetime.now()
+            contexts = await self.retriever.retrieve(query, k=k)
+            retrieval_time = (datetime.now() - start_time).total_seconds()
+            
+            self.logger.info(f"Retrieved {len(contexts.documents) if hasattr(contexts, 'documents') else len(contexts)} contexts in {retrieval_time:.2f}s")
+            return contexts
+            
+        except Exception as e:
+            self.logger.error(f"Error during retrieval: {e}")
+            raise
+    
+    async def rerank_contexts(self, 
+                            contexts: RetrievalResult, 
+                            query: str,
+                            top_k: Optional[int] = None) -> RetrievalResult:
+        """
+        Rerank retrieved contexts
+        
+        Args:
+            contexts: Retrieved contexts
+            query: Original query
+            top_k: Number of top contexts to keep after reranking
+            
+        Returns:
+            Reranked RetrievalResult object
+        """
+        if not self.reranker or not self.config.enable_reranking:
+            self.logger.info("Reranking disabled or reranker not available")
+            return contexts
+        
+        top_k = top_k or self.config.reranker_top_k
+        self.logger.info(f"Reranking contexts, keeping top {top_k}")
+        
+        try:
+            start_time = datetime.now()
+            reranked_contexts = await self.reranker.rerank(
+                query=query,
+                contexts=contexts,
+                top_k=top_k
+            )
+            rerank_time = (datetime.now() - start_time).total_seconds()
+            
+            self.logger.info(f"Reranking completed in {rerank_time:.2f}s")
+            return reranked_contexts
+            
+        except Exception as e:
+            self.logger.error(f"Error during reranking: {e}")
+            # Return original contexts if reranking fails
+            return contexts
+    
+    async def generate_response(self, 
+                              contexts: RetrievalResult, 
+                              query: Union[str, List[str]], 
+                              response_type: Union[List[str], str] = None,
+                              template_types: Optional[List[str]] = None,
+                              max_new_tokens: Optional[int] = None,
+                              **generation_kwargs) -> List[Dict[str, Any]]:
+        """
+        Generate responses based on contexts and query
+        
+        Args:
+            contexts: Retrieved contexts
+            query: User query or list of queries
+            response_type: Type(s) of response to generate
+            template_types: Template types for multi_response
+            max_new_tokens: Maximum tokens to generate
+            **generation_kwargs: Additional generation parameters
+            
+        Returns:
+            List of response dictionaries
+        """
+        await self._ensure_model_loaded()
+        
+        # Default response types
+        if response_type is None:
+            response_type = ["rag_response"]
+        elif isinstance(response_type, str):
+            response_type = [response_type]
+        
+        # Default template types
+        if template_types is None:
+            template_types = self.config.default_template_types
+        
+        responses = []
+        
+        try:
+            # RAG Response
+            if "rag_response" in response_type:
+                self.logger.info("Generating RAG response...")
+                start_time = datetime.now()
+                
+                if isinstance(query, list):
+                    # Handle multiple queries
+                    rag_responses = {}
+                    for i, q in enumerate(query):
+                        rag_response = await self.model.rag_generate(
+                            question=q,
+                            contexts=contexts,
+                            template_type="friendly",
+                            max_new_tokens=max_new_tokens,
+                            **generation_kwargs
+                        )
+                        rag_responses[f"query_{i}"] = rag_response
+                    responses.append({"rag_response": rag_responses})
+                else:
+                    rag_response = await self.model.rag_generate(
+                        question=query,
+                        contexts=contexts,
+                        template_type="friendly",
+                        max_new_tokens=max_new_tokens,
+                        **generation_kwargs
+                    )
+                    responses.append({"rag_response": rag_response})
+                
+                generation_time = (datetime.now() - start_time).total_seconds()
+                self.logger.info(f"RAG response generated in {generation_time:.2f}s")
+            
+            # Multi-template Response
+            if "multi_response" in response_type:
+                self.logger.info("Generating multi-template responses...")
+                start_time = datetime.now()
+                
+                if isinstance(query, list):
+                    multi_responses = {}
+                    for i, q in enumerate(query):
+                        multi_response = await self.model.multi_template_generate(
+                            question=q,
+                            contexts=contexts,
+                            template_types=template_types,
+                            max_new_tokens=max_new_tokens,
+                            **generation_kwargs
+                        )
+                        multi_responses[f"query_{i}"] = multi_response
+                    responses.append({"multi_responses": multi_responses})
+                else:
+                    multi_responses = await self.model.multi_template_generate(
+                        question=query,
+                        contexts=contexts,
+                        template_types=template_types,
+                        max_new_tokens=max_new_tokens,
+                        **generation_kwargs
+                    )
+                    responses.append({"multi_responses": multi_responses})
+                
+                generation_time = (datetime.now() - start_time).total_seconds()
+                self.logger.info(f"Multi-template responses generated in {generation_time:.2f}s")
+            
+            # Batch Response (untuk multiple prompts tanpa RAG context)
+            if "batch_response" in response_type:
+                self.logger.info("Generating batch responses...")
+                start_time = datetime.now()
+                
+                if isinstance(query, list):
+                    batch_responses = await self.model.batch_generate(
+                        query, 
+                        max_new_tokens=max_new_tokens,
+                        **generation_kwargs
+                    )
+                else:
+                    batch_responses = await self.model.batch_generate(
+                        [query], 
+                        max_new_tokens=max_new_tokens,
+                        **generation_kwargs
+                    )
+                
+                responses.append({"batch_responses": batch_responses})
+                
+                generation_time = (datetime.now() - start_time).total_seconds()
+                self.logger.info(f"Batch responses generated in {generation_time:.2f}s")
+            
+            return responses
+            
+        except Exception as e:
+            self.logger.error(f"Error during response generation: {e}")
+            raise
+    
+    async def generate_response_stream(self, 
+                                     contexts: RetrievalResult, 
+                                     query: str,
+                                     template_type: str = "main_template",
+                                     max_new_tokens: Optional[int] = None,
+                                     **generation_kwargs) -> AsyncGenerator[str, None]:
+        """
+        Generate RAG response with streaming
+        
+        Args:
+            contexts: Retrieved contexts
+            query: User query
+            template_type: Template type to use
+            max_new_tokens: Maximum tokens to generate
+            **generation_kwargs: Additional generation parameters
+            
+        Yields:
+            Response chunks
+        """
+        await self._ensure_model_loaded()
+        
+        self.logger.info(f"Generating streaming RAG response with template: {template_type}")
+        
+        async for chunk in self.model.rag_generate_stream(
+            question=query,
+            contexts=contexts,
+            template_type=template_type,
+            max_new_tokens=max_new_tokens,
+            **generation_kwargs
+        ):
+            yield chunk
+    
+    async def infer(self, 
+                   query: Union[str, List[str]], 
+                   response_type: Union[List[str], str] = None,
+                   k: Optional[int] = None,
+                   enable_reranking: Optional[bool] = None,
+                   template_types: Optional[List[str]] = None,
+                   max_new_tokens: Optional[int] = None,
+                   **generation_kwargs) -> Dict[str, Any]:
+        """
+        Complete inference pipeline
+        
+        Args:
+            query: User query or list of queries
+            response_type: Type(s) of response to generate
+            k: Number of contexts to retrieve
+            enable_reranking: Whether to enable reranking
+            template_types: Template types for multi_response
+            max_new_tokens: Maximum tokens to generate
+            **generation_kwargs: Additional generation parameters
+            
+        Returns:
+            Dictionary with results and metadata
+        """
+        start_time = datetime.now()
+        
+        # Handle single query
+        main_query = query[0] if isinstance(query, list) else query
+        
+        try:
+            # Step 1: Retrieve contexts
+            retrieved_contexts = await self.retrieve_context(main_query, k=k)
+            
+            # Step 2: Rerank contexts (if enabled)
+            enable_rerank = enable_reranking if enable_reranking is not None else self.config.enable_reranking
+            if enable_rerank:
+                contexts = await self.rerank_contexts(retrieved_contexts, main_query)
+            else:
+                contexts = retrieved_contexts
+            
+            # Step 3: Generate responses
+            responses = await self.generate_response(
+                contexts=contexts,
+                query=query,
+                response_type=response_type,
+                template_types=template_types,
+                max_new_tokens=max_new_tokens,
+                **generation_kwargs
+            )
+            
+            total_time = (datetime.now() - start_time).total_seconds()
+            
+            # Prepare result
+            result = {
+                "query": query,
+                "responses": responses,
+                "contexts": contexts,
+                "metadata": {
+                    "total_time": total_time,
+                    "retrieval_enabled": True,
+                    "reranking_enabled": enable_rerank,
+                    "num_contexts": len(contexts.documents) if hasattr(contexts, 'documents') else len(contexts),
+                    "response_types": response_type,
+                    "timestamp": datetime.now().isoformat()
+                }
+            }
+            
+            self.logger.info(f"Inference completed in {total_time:.2f}s")
+            return result
+            
+        except Exception as e:
+            self.logger.error(f"Error during inference: {e}")
+            raise
+    
+    async def infer_stream(self, 
+                          query: str,
+                          k: Optional[int] = None,
+                          enable_reranking: Optional[bool] = None,
+                          template_type: str = "main_template",
+                          max_new_tokens: Optional[int] = None,
+                          **generation_kwargs) -> AsyncGenerator[Dict[str, Any], None]:
+        """
+        Complete inference pipeline with streaming response
+        
+        Args:
+            query: User query
+            k: Number of contexts to retrieve
+            enable_reranking: Whether to enable reranking
+            template_type: Template type to use
+            max_new_tokens: Maximum tokens to generate
+            **generation_kwargs: Additional generation parameters
+            
+        Yields:
+            Dictionaries with stream data and metadata
+        """
+        start_time = datetime.now()
+        
+        try:
+            # Step 1: Retrieve contexts
+            retrieved_contexts = await self.retrieve_context(query, k=k)
+            
+            # Step 2: Rerank contexts (if enabled)
+            enable_rerank = enable_reranking if enable_reranking is not None else self.config.enable_reranking
+            if enable_rerank:
+                contexts = await self.rerank_contexts(retrieved_contexts, query)
+            else:
+                contexts = retrieved_contexts
+            
+            # Yield metadata first
+            setup_time = (datetime.now() - start_time).total_seconds()
+            yield {
+                "type": "metadata",
+                "data": {
+                    "query": query,
+                    "setup_time": setup_time,
+                    "num_contexts": len(contexts.documents) if hasattr(contexts, 'documents') else len(contexts),
+                    "reranking_enabled": enable_rerank,
+                    "template_type": template_type
+                }
+            }
+            
+            # Step 3: Stream response
+            response_start = datetime.now()
+            accumulated_text = ""
+            
+            async for chunk in self.generate_response_stream(
+                contexts=contexts,
+                query=query,
+                template_type=template_type,
+                max_new_tokens=max_new_tokens,
+                **generation_kwargs
+            ):
+                accumulated_text += chunk
+                yield {
+                    "type": "chunk",
+                    "data": {
+                        "chunk": chunk,
+                        "accumulated_text": accumulated_text,
+                        "generation_time": (datetime.now() - response_start).total_seconds()
+                    }
+                }
+            
+            # Yield final metadata
+            total_time = (datetime.now() - start_time).total_seconds()
+            yield {
+                "type": "complete",
+                "data": {
+                    "total_time": total_time,
+                    "final_response": accumulated_text,
+                    "contexts": contexts
+                }
+            }
+            
+        except Exception as e:
+            self.logger.error(f"Error during streaming inference: {e}")
+            yield {
+                "type": "error",
+                "data": {
+                    "error": str(e),
+                    "error_time": (datetime.now() - start_time).total_seconds()
+                }
+            }
+    
+    async def batch_infer(self, 
+                         queries: List[str],
+                         response_type: Union[List[str], str] = None,
+                         k: Optional[int] = None,
+                         enable_reranking: Optional[bool] = None,
+                         **generation_kwargs) -> List[Dict[str, Any]]:
+        """
+        Batch inference untuk multiple queries
+        
+        Args:
+            queries: List of queries
+            response_type: Type(s) of response to generate
+            k: Number of contexts to retrieve per query
+            enable_reranking: Whether to enable reranking
+            **generation_kwargs: Additional generation parameters
+            
+        Returns:
+            List of inference results
+        """
+        self.logger.info(f"Starting batch inference for {len(queries)} queries")
+        
+        # Create tasks untuk concurrent processing
+        tasks = [
+            asyncio.create_task(
+                self.infer(
+                    query=query,
+                    response_type=response_type,
+                    k=k,
+                    enable_reranking=enable_reranking,
+                    **generation_kwargs
+                )
+            )
+            for query in queries
+        ]
+        
+        # Wait for all tasks
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+        
+        # Process results
+        processed_results = []
+        for i, result in enumerate(results):
+            if isinstance(result, Exception):
+                self.logger.error(f"Error processing query {i}: {result}")
+                processed_results.append({
+                    "query": queries[i],
+                    "error": str(result),
+                    "success": False
+                })
+            else:
+                result["success"] = True
+                processed_results.append(result)
+        
+        return processed_results
+    
+    async def get_available_templates(self) -> List[str]:
+        """Get available template types from model"""
+        await self._ensure_model_loaded()
+        return self.model.get_available_templates()
+    
+    async def preview_template(self, 
+                              template_type: str, 
+                              sample_query: str = "Apa itu AI?") -> str:
+        """Preview template formatting"""
+        await self._ensure_model_loaded()
+        return self.model.preview_template(
+            template_type=template_type,
+            sample_question=sample_query,
+            sample_context="Sample context untuk preview template..."
+        )
+    
+    async def get_model_info(self) -> Dict[str, Any]:
+        """Get model information"""
+        await self._ensure_model_loaded()
+        model_info = await self.model.get_model_info()
+        
+        return {
+            "model_info": model_info,
+            "inferencer_config": self.config.__dict__,
+            "reranker_available": self.reranker is not None,
+            "available_templates": await self.get_available_templates()
+        }
+    
+    async def close(self):
+        """Clean up resources"""
+        self.logger.info("Closing Inferencer...")
+        if self.model:
+            await self.model.close()
+        self.logger.info("Inferencer closed successfully")
+    
+    async def __aenter__(self):
+        """Async context manager entry"""
+        await self._ensure_model_loaded()
+        return self
+    
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        """Async context manager exit"""
+        await self.close()